feat: handle CI success event auto trigger in batch

pkg/pipeline/CiConfig.go

@@ -86,6 +86,7 @@ type CiConfig struct {
 	OrchestratorToken                string `env:"ORCH_TOKEN" envDefault:""`
 	BuildxK8sDriverOptions           string `env:"BUILDX_K8S_DRIVER_OPTIONS" envDefault:""`
 	BuildxProvenanceMode             string `env:"BUILDX_PROVENANCE_MODE" envDefault:""` //provenance is set to false if this flag is not set
+	CIAutoTriggerBatchSize           int    `env:"CI_SUCCESS_AUTO_TRIGGER_BATCH_SIZE" envDefault:"1"`
 }
 
 type CiVolumeMount struct {

pkg/pipeline/WebhookService.go

@@ -34,6 +34,7 @@ import (
 	"go.uber.org/zap"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 )
 
@@ -58,6 +59,7 @@ type WebhookService interface {
 
 type WebhookServiceImpl struct {
 	ciArtifactRepository repository.CiArtifactRepository
+	ciConfig             *CiConfig
 	logger               *zap.SugaredLogger
 	ciPipelineRepository pipelineConfig.CiPipelineRepository
 	ciWorkflowRepository pipelineConfig.CiWorkflowRepository
@@ -70,6 +72,7 @@ type WebhookServiceImpl struct {
 
 func NewWebhookServiceImpl(
 	ciArtifactRepository repository.CiArtifactRepository,
+	ciConfig *CiConfig,
 	logger *zap.SugaredLogger,
 	ciPipelineRepository pipelineConfig.CiPipelineRepository,
 	appService app.AppService, eventClient client.EventClient,
@@ -78,6 +81,7 @@ func NewWebhookServiceImpl(
 	workflowDagExecutor WorkflowDagExecutor, ciHandler CiHandler) *WebhookServiceImpl {
 	return &WebhookServiceImpl{
 		ciArtifactRepository: ciArtifactRepository,
+		ciConfig:             ciConfig,
 		logger:               logger,
 		ciPipelineRepository: ciPipelineRepository,
 		appService:           appService,
@@ -239,13 +243,40 @@ func (impl WebhookServiceImpl) HandleCiSuccessEvent(ciPipelineId int, request *C
 		impl.logger.Debugw("Trigger (manual) by user", "userId", request.UserId)
 	}
 	async := false
-	for _, ciArtifact := range ciArtifactArr {
-		err = impl.workflowDagExecutor.HandleCiSuccessEvent(ciArtifact, isCiManual, async, request.UserId)
-		if err != nil {
-			impl.logger.Errorw("error on handle  ci success event", "err", err)
-			return 0, err
+
+	// execute auto trigger in batch on CI success event
+	totalCIArtifactCount := len(ciArtifactArr)
+	batchSize := impl.ciConfig.CIAutoTriggerBatchSize
+	// handling to avoid infinite loop
+	if batchSize <= 0 {
+		batchSize = 1
+	}
+	start := time.Now()
+	impl.logger.Infow("Started: auto trigger for children Stage/CD pipelines", "Artifact count", totalCIArtifactCount)
+	for i := 0; i < totalCIArtifactCount; {
+		//requests left to process
+		remainingBatch := totalCIArtifactCount - i
+		if remainingBatch < batchSize {
+			batchSize = remainingBatch
+		}
+		var wg sync.WaitGroup
+		for j := 0; j < batchSize; j++ {
+			wg.Add(1)
+			index := i + j
+			go func(index int) {
+				defer wg.Done()
+				ciArtifact := ciArtifactArr[index]
+				// handle individual CiArtifact success event
+				err = impl.workflowDagExecutor.HandleCiSuccessEvent(ciArtifact, isCiManual, async, request.UserId)
+				if err != nil {
+					impl.logger.Errorw("error on handle  ci success event", "ciArtifactId", ciArtifact.Id, "err", err)
+				}
+			}(index)
 		}
+		wg.Wait()
+		i += batchSize
 	}
+	impl.logger.Debugw("Completed: auto trigger for children Stage/CD pipelines", "Time taken", time.Since(start).Seconds())
 	return artifact.Id, err
 }