Remove redundant workflow status update logic for CI/CD workflows and refactor method signatures and repository integration in WorkflowStatusLatestService.

Author: prakash100198

pkg/pipeline/CiHandler.go

@@ -606,18 +606,6 @@ func (impl *CiHandlerImpl) UpdateWorkflow(workflowStatus eventProcessorBean.CiCd
 			return savedWorkflow.Id, true, err
 		}
 
-		// Update latest status table for CI workflow
-		// Check if CiPipeline is loaded, if not pass 0 as appId to let the function fetch it
-		appId := 0
-		if savedWorkflow.CiPipeline != nil {
-			appId = savedWorkflow.CiPipeline.AppId
-		}
-		err = impl.workflowStatusUpdateService.UpdateCiWorkflowStatusLatest(nil, savedWorkflow.CiPipelineId, appId, savedWorkflow.Id, savedWorkflow.TriggeredBy)
-		if err != nil {
-			impl.Logger.Errorw("error in updating ci workflow status latest", "err", err, "pipelineId", savedWorkflow.CiPipelineId, "workflowId", savedWorkflow.Id)
-			// Don't return error here as the main workflow update was successful
-		}
-
 		impl.sendCIFailEvent(savedWorkflow, status, message)
 		return savedWorkflow.Id, true, nil
 	}

pkg/pipeline/CiService.go

@@ -135,13 +135,7 @@ func (impl *CiServiceImpl) SaveCiWorkflowWithStage(wf *pipelineConfig.CiWorkflow
 		return err
 	}
 
-	// Update latest status table for CI workflow within the transaction
-	// Check if CiPipeline is loaded, if not pass 0 as appId to let the function fetch it
-	appId := 0
-	if wf.CiPipeline != nil {
-		appId = wf.CiPipeline.AppId
-	}
-	err = impl.workflowStatusUpdateService.UpdateCiWorkflowStatusLatest(tx, wf.CiPipelineId, appId, wf.Id, wf.TriggeredBy)
+	err = impl.workflowStatusUpdateService.SaveCiWorkflowStatusLatest(tx, wf.CiPipelineId, wf.Id, wf.TriggeredBy)
 	if err != nil {
 		impl.Logger.Errorw("error in updating ci workflow status latest", "err", err, "pipelineId", wf.CiPipelineId, "workflowId", wf.Id)
 		return err
@@ -184,18 +178,6 @@ func (impl *CiServiceImpl) UpdateCiWorkflowWithStage(wf *pipelineConfig.CiWorkfl
 		return err
 	}
 
-	// Update latest status table for CI workflow within the transaction
-	// Check if CiPipeline is loaded, if not pass 0 as appId to let the function fetch it
-	appId := 0
-	if wf.CiPipeline != nil {
-		appId = wf.CiPipeline.AppId
-	}
-	err = impl.workflowStatusUpdateService.UpdateCiWorkflowStatusLatest(tx, wf.CiPipelineId, appId, wf.Id, wf.TriggeredBy)
-	if err != nil {
-		impl.Logger.Errorw("error in updating ci workflow status latest", "err", err, "pipelineId", wf.CiPipelineId, "workflowId", wf.Id)
-		return err
-	}
-
 	err = impl.transactionManager.CommitTx(tx)
 	if err != nil {
 		impl.Logger.Errorw("error in committing transaction", "workflowName", wf.Name, "error", err)

pkg/workflow/cd/CdWorkflowRunnerService.go

@@ -80,21 +80,6 @@ func (impl *CdWorkflowRunnerServiceImpl) UpdateWfr(dto *bean.CdWorkflowRunnerDto
 		impl.logger.Errorw("error in updating runner status in db", "runnerId", runnerDbObj.Id, "err", err)
 		return err
 	}
-
-	// Update latest status table for CD workflow
-	// Check if CdWorkflow and Pipeline are loaded, if not pass 0 as appId/environmentId to let the function fetch them
-	appId := 0
-	environmentId := 0
-	if runnerDbObj.CdWorkflow != nil && runnerDbObj.CdWorkflow.Pipeline != nil {
-		appId = runnerDbObj.CdWorkflow.Pipeline.AppId
-		environmentId = runnerDbObj.CdWorkflow.Pipeline.EnvironmentId
-	}
-	err = impl.workflowStatusUpdateService.UpdateCdWorkflowStatusLatest(nil, runnerDbObj.CdWorkflow.PipelineId, appId, environmentId, runnerDbObj.Id, runnerDbObj.WorkflowType.String(), int32(updatedBy))
-	if err != nil {
-		impl.logger.Errorw("error in updating cd workflow status latest", "err", err, "pipelineId", runnerDbObj.CdWorkflow.PipelineId, "workflowRunnerId", runnerDbObj.Id)
-		// Don't return error here as the main workflow update was successful
-	}
-
 	return nil
 }
 

pkg/workflow/workflowStatusLatest/WorkflowStatusLatestService.go

@@ -30,12 +30,12 @@ import (
 
 type WorkflowStatusLatestService interface {
 	// CI Workflow Status Latest methods
-	SaveOrUpdateCiWorkflowStatusLatest(tx *pg.Tx, pipelineId, appId, ciWorkflowId int, userId int32) error
+	SaveCiWorkflowStatusLatest(tx *pg.Tx, pipelineId, ciWorkflowId int, userId int32) error
 	GetCiWorkflowStatusLatestByPipelineId(pipelineId int) (*CiWorkflowStatusLatest, error)
 	GetCiWorkflowStatusLatestByAppId(appId int) ([]*CiWorkflowStatusLatest, error)
 
 	// CD Workflow Status Latest methods
-	SaveOrUpdateCdWorkflowStatusLatest(tx *pg.Tx, pipelineId, appId, environmentId, workflowRunnerId int, workflowType string, userId int32) error
+	SaveCdWorkflowStatusLatest(tx *pg.Tx, pipelineId, appId, environmentId, workflowRunnerId int, workflowType string, userId int32) error
 	GetCdWorkflowStatusLatestByPipelineIdAndWorkflowType(pipelineId int, workflowType string) (*CdWorkflowStatusLatest, error)
 	GetCdWorkflowStatusLatestByAppId(appId int) ([]*CdWorkflowStatusLatest, error)
 	GetCdWorkflowStatusLatestByPipelineId(pipelineId int) ([]*CdWorkflowStatusLatest, error)
@@ -47,19 +47,22 @@ type WorkflowStatusLatestServiceImpl struct {
 	workflowStatusLatestRepository pipelineConfig.WorkflowStatusLatestRepository
 	ciWorkflowRepository           pipelineConfig.CiWorkflowRepository
 	cdWorkflowRepository           pipelineConfig.CdWorkflowRepository
+	ciPipelineRepository           pipelineConfig.CiPipelineRepository
 }
 
 func NewWorkflowStatusLatestServiceImpl(
 	logger *zap.SugaredLogger,
 	workflowStatusLatestRepository pipelineConfig.WorkflowStatusLatestRepository,
 	ciWorkflowRepository pipelineConfig.CiWorkflowRepository,
 	cdWorkflowRepository pipelineConfig.CdWorkflowRepository,
+	ciPipelineRepository pipelineConfig.CiPipelineRepository,
 ) *WorkflowStatusLatestServiceImpl {
 	return &WorkflowStatusLatestServiceImpl{
 		logger:                         logger,
 		workflowStatusLatestRepository: workflowStatusLatestRepository,
 		ciWorkflowRepository:           ciWorkflowRepository,
 		cdWorkflowRepository:           cdWorkflowRepository,
+		ciPipelineRepository:           ciPipelineRepository,
 	}
 }
 
@@ -80,8 +83,7 @@ type CdWorkflowStatusLatest struct {
 	Status           string `json:"status"` // Derived from cd_workflow_runner table
 }
 
-// CI Workflow Status Latest methods implementation
-func (impl *WorkflowStatusLatestServiceImpl) SaveOrUpdateCiWorkflowStatusLatest(tx *pg.Tx, pipelineId, appId, ciWorkflowId int, userId int32) error {
+func (impl *WorkflowStatusLatestServiceImpl) SaveCiWorkflowStatusLatest(tx *pg.Tx, pipelineId, ciWorkflowId int, userId int32) error {
 	// Validate required parameters
 	if pipelineId <= 0 {
 		impl.logger.Errorw("invalid pipelineId provided", "pipelineId", pipelineId)
@@ -92,44 +94,22 @@ func (impl *WorkflowStatusLatestServiceImpl) SaveOrUpdateCiWorkflowStatusLatest(
 		impl.logger.Errorw("invalid ciWorkflowId provided", "ciWorkflowId", ciWorkflowId)
 		return fmt.Errorf("invalid ciWorkflowId: %d", ciWorkflowId)
 	}
-
-	// If appId is not provided (0), fetch it from the CiPipeline
-	if appId <= 0 {
-		ciPipeline, err := impl.ciWorkflowRepository.FindById(ciWorkflowId)
-		if err != nil {
-			impl.logger.Errorw("error in fetching ci workflow to get appId", "err", err, "ciWorkflowId", ciWorkflowId)
-			return err
-		}
-
-		if ciPipeline == nil {
-			impl.logger.Errorw("ci workflow not found", "ciWorkflowId", ciWorkflowId)
-			return fmt.Errorf("ci workflow not found with id: %d", ciWorkflowId)
-		}
-
-		// Check if CiPipeline is loaded
-		if ciPipeline.CiPipeline == nil {
-			impl.logger.Errorw("ci pipeline not loaded in ci workflow", "ciWorkflowId", ciWorkflowId, "ciPipelineId", ciPipeline.CiPipelineId)
-			return fmt.Errorf("ci pipeline not loaded for workflow id: %d", ciWorkflowId)
-		}
-
-		appId = ciPipeline.CiPipeline.AppId
-		if appId <= 0 {
-			impl.logger.Errorw("invalid appId in ci pipeline", "ciWorkflowId", ciWorkflowId, "ciPipelineId", ciPipeline.CiPipelineId, "appId", appId)
-			return fmt.Errorf("invalid appId in ci pipeline: %d", appId)
-		}
-
-		impl.logger.Debugw("fetched appId from ci workflow", "ciWorkflowId", ciWorkflowId, "appId", appId)
+	ciPipeline, err := impl.ciPipelineRepository.FindOneWithAppData(pipelineId)
+	if err != nil {
+		impl.logger.Errorw("error in fetching ci pipeline for appId", "ciPipelineId", pipelineId, "err", err)
+		return err
 	}
+	appId := ciPipeline.AppId
 
 	// Check if entry exists
 	existingEntry, err := impl.workflowStatusLatestRepository.GetCiWorkflowStatusLatestByPipelineId(pipelineId)
-	if err != nil && err != pg.ErrNoRows {
+	if err != nil && !util2.IsErrNoRows(err) {
 		impl.logger.Errorw("error in getting ci workflow status latest", "err", err, "pipelineId", pipelineId)
 		return err
 	}
 
 	now := time.Now()
-	if err == pg.ErrNoRows {
+	if util2.IsErrNoRows(err) {
 		// Create new entry
 		model := &pipelineConfig.CiWorkflowStatusLatest{
 			PipelineId:   pipelineId,
@@ -207,7 +187,7 @@ func (impl *WorkflowStatusLatestServiceImpl) GetCiWorkflowStatusLatestByAppId(ap
 }
 
 // CD Workflow Status Latest methods implementation
-func (impl *WorkflowStatusLatestServiceImpl) SaveOrUpdateCdWorkflowStatusLatest(tx *pg.Tx, pipelineId, appId, environmentId, workflowRunnerId int, workflowType string, userId int32) error {
+func (impl *WorkflowStatusLatestServiceImpl) SaveCdWorkflowStatusLatest(tx *pg.Tx, pipelineId, appId, environmentId, workflowRunnerId int, workflowType string, userId int32) error {
 	// Validate required parameters
 	if pipelineId <= 0 {
 		impl.logger.Errorw("invalid pipelineId provided", "pipelineId", pipelineId)

pkg/workflow/workflowStatusLatest/WorkflowStatusUpdateService.go

@@ -24,12 +24,8 @@ import (
 
 type WorkflowStatusUpdateService interface {
 	// Methods to update latest status tables when workflow status changes
-	UpdateCiWorkflowStatusLatest(tx *pg.Tx, pipelineId, appId, ciWorkflowId int, userId int32) error
+	SaveCiWorkflowStatusLatest(tx *pg.Tx, pipelineId, ciWorkflowId int, userId int32) error
 	UpdateCdWorkflowStatusLatest(tx *pg.Tx, pipelineId, appId, environmentId, workflowRunnerId int, workflowType string, userId int32) error
-
-	// Methods to fetch optimized status for trigger view
-	FetchCiStatusForTriggerViewOptimized(appId int) ([]*pipelineConfig.CiWorkflowStatus, error)
-	FetchCdStatusForTriggerViewOptimized(appId int) ([]*pipelineConfig.CdWorkflowStatus, error)
 }
 
 type WorkflowStatusUpdateServiceImpl struct {
@@ -59,101 +55,10 @@ func NewWorkflowStatusUpdateServiceImpl(
 	}
 }
 
-func (impl *WorkflowStatusUpdateServiceImpl) UpdateCiWorkflowStatusLatest(tx *pg.Tx, pipelineId, appId, ciWorkflowId int, userId int32) error {
-	return impl.workflowStatusLatestService.SaveOrUpdateCiWorkflowStatusLatest(tx, pipelineId, appId, ciWorkflowId, userId)
+func (impl *WorkflowStatusUpdateServiceImpl) SaveCiWorkflowStatusLatest(tx *pg.Tx, pipelineId, ciWorkflowId int, userId int32) error {
+	return impl.workflowStatusLatestService.SaveCiWorkflowStatusLatest(tx, pipelineId, ciWorkflowId, userId)
 }
 
 func (impl *WorkflowStatusUpdateServiceImpl) UpdateCdWorkflowStatusLatest(tx *pg.Tx, pipelineId, appId, environmentId, workflowRunnerId int, workflowType string, userId int32) error {
-	return impl.workflowStatusLatestService.SaveOrUpdateCdWorkflowStatusLatest(tx, pipelineId, appId, environmentId, workflowRunnerId, workflowType, userId)
-}
-
-func (impl *WorkflowStatusUpdateServiceImpl) FetchCiStatusForTriggerViewOptimized(appId int) ([]*pipelineConfig.CiWorkflowStatus, error) {
-	latestStatuses, err := impl.workflowStatusLatestService.GetCiWorkflowStatusLatestByAppId(appId)
-	if err != nil {
-		impl.logger.Errorw("error in getting ci workflow status latest by app id", "err", err, "appId", appId)
-		// Fallback to old method
-		return impl.ciWorkflowRepository.FIndCiWorkflowStatusesByAppId(appId)
-	}
-
-	// Convert to the expected format
-	var ciWorkflowStatuses []*pipelineConfig.CiWorkflowStatus
-	for _, latestStatus := range latestStatuses {
-		ciPipeline, err := impl.ciPipelineRepository.FindById(latestStatus.PipelineId)
-		if err != nil {
-			impl.logger.Errorw("error in getting ci pipeline", "err", err, "pipelineId", latestStatus.PipelineId)
-			continue
-		}
-
-		ciWorkflowStatus := &pipelineConfig.CiWorkflowStatus{
-			CiPipelineId:      latestStatus.PipelineId,
-			CiPipelineName:    ciPipeline.Name,
-			CiStatus:          latestStatus.Status,
-			CiWorkflowId:      latestStatus.CiWorkflowId,
-			StorageConfigured: latestStatus.StorageConfigured,
-		}
-		ciWorkflowStatuses = append(ciWorkflowStatuses, ciWorkflowStatus)
-	}
-
-	// If no entries found in latest status table, fallback to old method
-	if len(ciWorkflowStatuses) == 0 {
-		impl.logger.Infow("no entries found in ci workflow status latest table, falling back to old method", "appId", appId)
-		return impl.ciWorkflowRepository.FIndCiWorkflowStatusesByAppId(appId)
-	}
-
-	return ciWorkflowStatuses, nil
-}
-
-func (impl *WorkflowStatusUpdateServiceImpl) FetchCdStatusForTriggerViewOptimized(appId int) ([]*pipelineConfig.CdWorkflowStatus, error) {
-	// First try to get from the optimized latest status table
-	latestStatuses, err := impl.workflowStatusLatestService.GetCdWorkflowStatusLatestByAppId(appId)
-	if err != nil {
-		impl.logger.Errorw("error in getting cd workflow status latest by app id", "err", err, "appId", appId)
-		// Fallback to old method - would need to implement this based on existing CD status fetching logic
-		return nil, err
-	}
-
-	// Convert to the expected format
-	var cdWorkflowStatuses []*pipelineConfig.CdWorkflowStatus
-	for _, latestStatus := range latestStatuses {
-		// Get pipeline info
-		pipeline, err := impl.pipelineRepository.FindById(latestStatus.PipelineId)
-		if err != nil {
-			impl.logger.Errorw("error in getting pipeline", "err", err, "pipelineId", latestStatus.PipelineId)
-			continue
-		}
-
-		var status string
-		switch latestStatus.WorkflowType {
-		case "PRE":
-			status = latestStatus.Status
-		case "DEPLOY":
-			status = latestStatus.Status
-		case "POST":
-			status = latestStatus.Status
-		default:
-			status = latestStatus.Status
-		}
-
-		cdWorkflowStatus := &pipelineConfig.CdWorkflowStatus{
-			CiPipelineId: pipeline.CiPipelineId,
-			PipelineId:   latestStatus.PipelineId,
-			PipelineName: pipeline.Name,
-			WorkflowType: latestStatus.WorkflowType,
-			WfrId:        latestStatus.WorkflowRunnerId,
-		}
-
-		// Set the appropriate status field based on workflow type
-		switch latestStatus.WorkflowType {
-		case "PRE":
-			cdWorkflowStatus.PreStatus = status
-		case "DEPLOY":
-			cdWorkflowStatus.DeployStatus = status
-		case "POST":
-			cdWorkflowStatus.PostStatus = status
-		}
-
-		cdWorkflowStatuses = append(cdWorkflowStatuses, cdWorkflowStatus)
-	}
-
-	return cdWorkflowStatuses, nil
+	return impl.workflowStatusLatestService.SaveCdWorkflowStatusLatest(tx, pipelineId, appId, environmentId, workflowRunnerId, workflowType, userId)
 }