Implement hybrid CI workflow status fetching in CiHandlerImpl using ci_workflow_status_latest table with fallback to legacy query methods. Update repositories with new helper methods to support this approach.

Author: prakash100198

internal/sql/repository/pipelineConfig/CiWorkflowRepository.go

@@ -40,7 +40,7 @@ type CiWorkflowRepository interface {
 	FindByName(name string) (*CiWorkflow, error)
 
 	FindLastTriggeredWorkflowByCiIds(pipelineId []int) (ciWorkflow []*CiWorkflow, err error)
-	FindLastTriggeredWorkflowByCiIdsOptimized(pipelineId []int) (ciWorkflow []*CiWorkflow, err error)
+	FindWorkflowsByCiWorkflowIds(ciWorkflowIds []int) (ciWorkflow []*CiWorkflow, err error)
 	FindLastTriggeredWorkflowByArtifactId(ciArtifactId int) (ciWorkflow *CiWorkflow, err error)
 	FindAllLastTriggeredWorkflowByArtifactId(ciArtifactId []int) (ciWorkflow []*CiWorkflow, err error)
 	FindAllTriggeredWorkflowCountInLast24Hour() (ciWorkflowCount int, err error)
@@ -49,6 +49,7 @@ type CiWorkflowRepository interface {
 	ExistsByStatus(status string) (bool, error)
 	FindBuildTypeAndStatusDataOfLast1Day() ([]*BuildTypeCount, error)
 	FIndCiWorkflowStatusesByAppId(appId int) ([]*CiWorkflowStatus, error)
+	FindCiPipelineIdsByAppId(appId int) ([]int, error)
 
 	MigrateIsArtifactUploaded(wfId int, isArtifactUploaded bool)
 	MigrateCiArtifactLocation(wfId int, artifactLocation string)
@@ -291,20 +292,16 @@ func (impl *CiWorkflowRepositoryImpl) FindLastTriggeredWorkflowByCiIds(pipelineI
 	return ciWorkflow, err
 }
 
-// FindLastTriggeredWorkflowByCiIdsOptimized uses the ci_workflow_status_latest table for better performance
-func (impl *CiWorkflowRepositoryImpl) FindLastTriggeredWorkflowByCiIdsOptimized(pipelineId []int) (ciWorkflow []*CiWorkflow, err error) {
+// FindWorkflowsByCiWorkflowIds fetches workflows by their workflow IDs (simple query)
+func (impl *CiWorkflowRepositoryImpl) FindWorkflowsByCiWorkflowIds(ciWorkflowIds []int) (ciWorkflow []*CiWorkflow, err error) {
+	if len(ciWorkflowIds) == 0 {
+		return ciWorkflow, nil
+	}
+
 	err = impl.dbConnection.Model(&ciWorkflow).
 		Column("ci_workflow.*", "CiPipeline").
-		Join("INNER JOIN ci_workflow_status_latest cwsl ON cwsl.ci_workflow_id = ci_workflow.id").
-		Where("cwsl.pipeline_id IN (?)", pg.In(pipelineId)).
+		Where("ci_workflow.id IN (?)", pg.In(ciWorkflowIds)).
 		Select()
-
-	if err != nil {
-		impl.logger.Errorw("error in optimized query, falling back to old method", "err", err, "pipelineIds", pipelineId)
-		// Fallback to the old method if optimized query fails
-		return impl.FindLastTriggeredWorkflowByCiIds(pipelineId)
-	}
-
 	return ciWorkflow, err
 }
 
@@ -397,6 +394,16 @@ func (impl *CiWorkflowRepositoryImpl) FIndCiWorkflowStatusesByAppId(appId int) (
 	return ciworkflowStatuses, err
 }
 
+// FindCiPipelineIdsByAppId gets all CI pipeline IDs for an app (simple query)
+func (impl *CiWorkflowRepositoryImpl) FindCiPipelineIdsByAppId(appId int) ([]int, error) {
+	var ciPipelineIds []int
+	err := impl.dbConnection.Model((*CiPipeline)(nil)).
+		Column("id").
+		Where("app_id = ? AND deleted = false", appId).
+		Select(&ciPipelineIds)
+	return ciPipelineIds, err
+}
+
 func (impl *CiWorkflowRepositoryImpl) MigrateIsArtifactUploaded(wfId int, isArtifactUploaded bool) {
 	_, err := impl.dbConnection.Model((*CiWorkflow)(nil)).
 		Set("is_artifact_uploaded = ?", workflow.GetArtifactUploadedType(isArtifactUploaded)).

internal/sql/repository/pipelineConfig/WorkflowStatusLatestRepository.go

@@ -28,6 +28,8 @@ type WorkflowStatusLatestRepository interface {
 	UpdateCiWorkflowStatusLatest(model *CiWorkflowStatusLatest) error
 	GetCiWorkflowStatusLatestByPipelineId(pipelineId int) (*CiWorkflowStatusLatest, error)
 	GetCiWorkflowStatusLatestByAppId(appId int) ([]*CiWorkflowStatusLatest, error)
+	GetCachedPipelineIds(pipelineIds []int) ([]int, error)
+	GetCiWorkflowStatusLatestByPipelineIds(pipelineIds []int) ([]*CiWorkflowStatusLatest, error)
 	DeleteCiWorkflowStatusLatestByPipelineId(pipelineId int) error
 
 	// CD Workflow Status Latest methods
@@ -127,6 +129,39 @@ func (impl *WorkflowStatusLatestRepositoryImpl) DeleteCiWorkflowStatusLatestByPi
 	return nil
 }
 
+func (impl *WorkflowStatusLatestRepositoryImpl) GetCachedPipelineIds(pipelineIds []int) ([]int, error) {
+	if len(pipelineIds) == 0 {
+		return []int{}, nil
+	}
+
+	var cachedPipelineIds []int
+	err := impl.dbConnection.Model(&CiWorkflowStatusLatest{}).
+		Column("pipeline_id").
+		Where("pipeline_id IN (?)", pg.In(pipelineIds)).
+		Select(&cachedPipelineIds)
+	if err != nil {
+		impl.logger.Errorw("error in getting cached pipeline ids", "err", err, "pipelineIds", pipelineIds)
+		return nil, err
+	}
+	return cachedPipelineIds, nil
+}
+
+func (impl *WorkflowStatusLatestRepositoryImpl) GetCiWorkflowStatusLatestByPipelineIds(pipelineIds []int) ([]*CiWorkflowStatusLatest, error) {
+	if len(pipelineIds) == 0 {
+		return []*CiWorkflowStatusLatest{}, nil
+	}
+
+	var models []*CiWorkflowStatusLatest
+	err := impl.dbConnection.Model(&models).
+		Where("pipeline_id IN (?)", pg.In(pipelineIds)).
+		Select()
+	if err != nil {
+		impl.logger.Errorw("error in getting ci workflow status latest by pipeline ids", "err", err, "pipelineIds", pipelineIds)
+		return nil, err
+	}
+	return models, nil
+}
+
 // CD Workflow Status Latest methods implementation
 func (impl *WorkflowStatusLatestRepositoryImpl) SaveCdWorkflowStatusLatest(model *CdWorkflowStatusLatest) error {
 	err := impl.dbConnection.Insert(model)