fix: [TKC-5111] use env id for queue/start workflow events (#7339)

Author: povilasv

cmd/api-server/services/controlplane.go

@@ -66,7 +66,7 @@ func CreateControlPlane(ctx context.Context, cfg *config.Config, eventsEmitter *
 	commands := controlplane.CreateCommands(cfg.StorageBucket, storageClient, testWorkflowOutputRepository, testWorkflowResultsRepository, artifactStorage)
 
 	enqueuer := scheduling.NewEnqueuer(log.DefaultLogger, testWorkflowsClient, testWorkflowTemplatesClient, testWorkflowResultsRepository, eventsEmitter,
-		cfg.GlobalWorkflowTemplateName, cfg.GlobalWorkflowTemplateInline != "")
+		envID, cfg.GlobalWorkflowTemplateName, cfg.GlobalWorkflowTemplateInline != "")
 	scheduler := factory.NewScheduler()
 	executionController := factory.NewExecutionController()
 	executionQuerier := factory.NewExecutionQuerier()

pkg/api/v1/testkube/model_event_extended.go

@@ -31,12 +31,7 @@ func NewEvent(t *EventType, resource *EventResource, id string) Event {
 	}
 }
 
-func NewEventQueueTestWorkflow(execution *TestWorkflowExecution) Event {
-	groupId := ""
-	if execution != nil {
-		groupId = execution.GroupId
-	}
-
+func NewEventQueueTestWorkflow(execution *TestWorkflowExecution, groupId string) Event {
 	return Event{
 		Id:                    uuid.NewString(),
 		GroupId:               groupId,
@@ -45,12 +40,7 @@ func NewEventQueueTestWorkflow(execution *TestWorkflowExecution) Event {
 	}
 }
 
-func NewEventStartTestWorkflow(execution *TestWorkflowExecution) Event {
-	groupId := ""
-	if execution != nil {
-		groupId = execution.GroupId
-	}
-
+func NewEventStartTestWorkflow(execution *TestWorkflowExecution, groupId string) Event {
 	return Event{
 		Id:                    uuid.NewString(),
 		GroupId:               groupId,

pkg/api/v1/testkube/model_event_extended_test.go

@@ -122,3 +122,15 @@ func TestEvent_IsSuccess(t *testing.T) {
 		}
 	})
 }
+
+func TestNewWorkflowEvents_UseExplicitRoutingGroup(t *testing.T) {
+	t.Parallel()
+
+	execution := &TestWorkflowExecution{Id: "execution-1", GroupId: "execution-group-1"}
+
+	queueEvent := NewEventQueueTestWorkflow(execution, "env-123")
+	startEvent := NewEventStartTestWorkflow(execution, "env-123")
+
+	assert.Equal(t, "env-123", queueEvent.GroupId)
+	assert.Equal(t, "env-123", startEvent.GroupId)
+}

pkg/controlplane/agent_grpc_execution_updates.go

@@ -77,8 +77,8 @@ func (s *Server) GetExecutionUpdates(ctx context.Context, _ *executionv1.GetExec
 				log.Warnw("error marking execution as starting", "err", err)
 			}
 
-			// Dispatch event for WebHooks and friends
-			s.emitter.Notify(testkube.NewEventStartTestWorkflow(&exe))
+			// Dispatch event for WebHooks and friends using the environment ID as the routing group.
+			s.emitter.Notify(testkube.NewEventStartTestWorkflow(&exe, s.envID))
 		default:
 			log.Warnw("unexpected state", "id", exe.Id, "status", *exe.Result.Status)
 		}

pkg/controlplane/scheduling/enqueuer.go

@@ -28,6 +28,7 @@ type Enqueuer struct {
 	templateRepository       testworkflowtemplateclient.TestWorkflowTemplateClient
 	executionRepository      testworkflow.Repository
 	emitter                  *event.Emitter
+	envID                    string
 	globalTemplateName       string
 	hasInlinedGlobalTemplate bool
 }
@@ -38,6 +39,7 @@ func NewEnqueuer(
 	templateRepository testworkflowtemplateclient.TestWorkflowTemplateClient,
 	executionRepository testworkflow.Repository,
 	emitter *event.Emitter,
+	envID string,
 	globalTemplateName string,
 	hasInlinedGlobalTemplate bool,
 ) Enqueuer {
@@ -47,6 +49,7 @@ func NewEnqueuer(
 		templateRepository:       templateRepository,
 		executionRepository:      executionRepository,
 		emitter:                  emitter,
+		envID:                    envID,
 		globalTemplateName:       globalTemplateName,
 		hasInlinedGlobalTemplate: hasInlinedGlobalTemplate,
 	}
@@ -394,7 +397,7 @@ func (e *Enqueuer) persistExecution(ctx context.Context, executions []*testworkf
 // dispatchExecutionEvents dispatches events related to queueing.
 func (e *Enqueuer) dispatchExecutionEvents(executions []testkube.TestWorkflowExecution) {
 	for _, execution := range executions {
-		e.emitter.Notify(testkube.NewEventQueueTestWorkflow(&execution))
+		e.emitter.Notify(testkube.NewEventQueueTestWorkflow(&execution, e.envID))
 	}
 }
 

pkg/event/bus/nats_integration_test.go

@@ -82,7 +82,7 @@ func TestNATS_Integration(t *testing.T) {
 
 	// given event
 
-	event := testkube.NewEventStartTestWorkflow(testkube.NewQueuedExecution())
+	event := testkube.NewEventStartTestWorkflow(testkube.NewQueuedExecution(), "")
 	event.Id = "123"
 
 	// and connection

pkg/event/emitter_test.go

@@ -215,10 +215,10 @@ func TestEmitter_GroupedListenersReceiveQueuedAndStartedWorkflowEvents(t *testin
 		time.Sleep(50 * time.Millisecond)
 
 		execution := testkube.NewExecutionWithID("executionID-grouped", "grouped")
-		execution.GroupId = "env-123"
+		execution.GroupId = "execution-group-123"
 
-		emitter.Notify(testkube.NewEventQueueTestWorkflow(execution))
-		emitter.Notify(testkube.NewEventStartTestWorkflow(execution))
+		emitter.Notify(testkube.NewEventQueueTestWorkflow(execution, "env-123"))
+		emitter.Notify(testkube.NewEventStartTestWorkflow(execution, "env-123"))
 
 		assert.Eventually(t, func() bool {
 			return listener.GetNotificationCount() == 2 && len(listener.GetReceivedEventTypes()) == 2

pkg/event/kind/websocket/listener_test.go

@@ -20,7 +20,7 @@ func TestWebsocketListener(t *testing.T) {
 	}}
 
 	// when
-	result := l.Notify(testkube.NewEventStartTestWorkflow(testkube.NewQueuedExecution()))
+	result := l.Notify(testkube.NewEventStartTestWorkflow(testkube.NewQueuedExecution(), ""))
 
 	// then
 	assert.Equal(t, "", result.Error_)
@@ -31,7 +31,7 @@ func TestWebsocketListenerNoClients(t *testing.T) {
 	l := NewWebsocketListener()
 
 	// when
-	result := l.Notify(testkube.NewEventStartTestWorkflow(testkube.NewQueuedExecution()))
+	result := l.Notify(testkube.NewEventStartTestWorkflow(testkube.NewQueuedExecution(), ""))
 
 	// then - not an error when no clients are connected
 	assert.Equal(t, "", result.Error_)