From ff3b8b6063b34082d0b07b119677002cad50246d Mon Sep 17 00:00:00 2001
From: Francisco Javier Tirado Sarti
 <65240126+fjtirado@users.noreply.github.com>
Date: Tue, 4 Jun 2024 12:51:55 +0200
Subject: [PATCH] [Fix_#3451] Fixing start event state filtering (#3542)

* [Fix_#3451] Fixing start event state filtering

* [Fix_#3451] Add IT test
---
 .../parser/handlers/EventHandler.java         |  3 +-
 .../parser/handlers/StateHandler.java         | 12 +++++-
 .../src/main/resources/application.properties |  6 +++
 .../src/main/resources/greetCustomer.sw.json  | 41 +++++++++++++++++++
 .../kogito/quarkus/workflows/EventFlowIT.java | 22 ++++++++++
 5 files changed, 80 insertions(+), 4 deletions(-)
 create mode 100644 quarkus/extensions/kogito-quarkus-serverless-workflow-extension/kogito-quarkus-serverless-workflow-integration-test/src/main/resources/greetCustomer.sw.json

diff --git a/kogito-serverless-workflow/kogito-serverless-workflow-builder/src/main/java/org/kie/kogito/serverless/workflow/parser/handlers/EventHandler.java b/kogito-serverless-workflow/kogito-serverless-workflow-builder/src/main/java/org/kie/kogito/serverless/workflow/parser/handlers/EventHandler.java
index 9ccecb22a79..6c95879a018 100644
--- a/kogito-serverless-workflow/kogito-serverless-workflow-builder/src/main/java/org/kie/kogito/serverless/workflow/parser/handlers/EventHandler.java
+++ b/kogito-serverless-workflow/kogito-serverless-workflow-builder/src/main/java/org/kie/kogito/serverless/workflow/parser/handlers/EventHandler.java
@@ -30,7 +30,6 @@
 import org.jbpm.workflow.core.node.Join;
 import org.jbpm.workflow.core.node.Split;
 import org.kie.kogito.serverless.workflow.parser.ParserContext;
-import org.kie.kogito.serverless.workflow.parser.ServerlessWorkflowParser;
 
 import io.serverlessworkflow.api.Workflow;
 import io.serverlessworkflow.api.events.OnEvents;
@@ -65,7 +64,7 @@ public MakeNodeResult makeNode(RuleFlowNodeContainerFactory<?, ?> factory) {
 
     private MakeNodeResult processOnEvent(RuleFlowNodeContainerFactory<?, ?> factory, OnEvents onEvent) {
         MakeNodeResult result = joinNodes(factory,
-                onEvent.getEventRefs(), (fact, onEventRef) -> filterAndMergeNode(fact, onEvent.getEventDataFilter(), isStartState ? ServerlessWorkflowParser.DEFAULT_WORKFLOW_VAR : getVarName(),
+                onEvent.getEventRefs(), (fact, onEventRef) -> filterAndMergeNode(fact, onEvent.getEventDataFilter(), getVarName(),
                         (f, inputVar, outputVar) -> buildEventNode(f, onEventRef, inputVar, outputVar)));
         CompositeContextNodeFactory<?> embeddedSubProcess = handleActions(makeCompositeNode(factory), onEvent.getActions());
         connect(result.getOutgoingNode(), embeddedSubProcess);
diff --git a/kogito-serverless-workflow/kogito-serverless-workflow-builder/src/main/java/org/kie/kogito/serverless/workflow/parser/handlers/StateHandler.java b/kogito-serverless-workflow/kogito-serverless-workflow-builder/src/main/java/org/kie/kogito/serverless/workflow/parser/handlers/StateHandler.java
index 63610ad6b71..5c9caf6f451 100644
--- a/kogito-serverless-workflow/kogito-serverless-workflow-builder/src/main/java/org/kie/kogito/serverless/workflow/parser/handlers/StateHandler.java
+++ b/kogito-serverless-workflow/kogito-serverless-workflow-builder/src/main/java/org/kie/kogito/serverless/workflow/parser/handlers/StateHandler.java
@@ -50,6 +50,7 @@
 import org.kie.kogito.serverless.workflow.SWFConstants;
 import org.kie.kogito.serverless.workflow.parser.ParserContext;
 import org.kie.kogito.serverless.workflow.parser.ServerlessWorkflowParser;
+import org.kie.kogito.serverless.workflow.parser.VariableInfo;
 import org.kie.kogito.serverless.workflow.suppliers.CollectorActionSupplier;
 import org.kie.kogito.serverless.workflow.suppliers.CompensationActionSupplier;
 import org.kie.kogito.serverless.workflow.suppliers.ErrorExpressionActionSupplier;
@@ -424,7 +425,8 @@ protected final MakeNodeResult filterAndMergeNode(RuleFlowNodeContainerFactory<?
             toExpr = eventFilter.getToStateData();
             useData = eventFilter.isUseData();
         }
-        return filterAndMergeNode(embeddedSubProcess, varName, null, dataExpr, toExpr, useData, true, nodeSupplier);
+        return filterAndMergeNode(embeddedSubProcess, isStartState ? new VariableInfo(DEFAULT_WORKFLOW_VAR, varName) : new VariableInfo(varName, varName), null, dataExpr, toExpr, useData, true,
+                nodeSupplier);
     }
 
     protected boolean isTempVariable(String varName) {
@@ -434,7 +436,13 @@ protected boolean isTempVariable(String varName) {
     protected final MakeNodeResult filterAndMergeNode(RuleFlowNodeContainerFactory<?, ?> embeddedSubProcess, String actionVarName, String fromStateExpr, String resultExpr, String toStateExpr,
             boolean useData,
             boolean shouldMerge, FilterableNodeSupplier nodeSupplier) {
+        return filterAndMergeNode(embeddedSubProcess, new VariableInfo(actionVarName, actionVarName), fromStateExpr, resultExpr, toStateExpr, useData, shouldMerge, nodeSupplier);
+    }
 
+    protected final MakeNodeResult filterAndMergeNode(RuleFlowNodeContainerFactory<?, ?> embeddedSubProcess, VariableInfo variableInfo, String fromStateExpr, String resultExpr, String toStateExpr,
+            boolean useData,
+            boolean shouldMerge, FilterableNodeSupplier nodeSupplier) {
+        String actionVarName = variableInfo.getOutputVar();
         if (isTempVariable(actionVarName)) {
             embeddedSubProcess.variable(actionVarName, new ObjectDataType(JsonNode.class.getCanonicalName()), Map.of(KogitoTags.VARIABLE_TAGS, KogitoTags.INTERNAL_TAG));
         }
@@ -451,7 +459,7 @@ protected final MakeNodeResult filterAndMergeNode(RuleFlowNodeContainerFactory<?
 
         if (useData && resultExpr != null) {
             currentNode = connect(currentNode, embeddedSubProcess.actionNode(parserContext.newId()).action(ExpressionActionSupplier.of(workflow, resultExpr)
-                    .withVarNames(actionVarName, actionVarName).build()));
+                    .withVarNames(variableInfo.getInputVar(), actionVarName).build()));
         }
 
         if (useData) {
diff --git a/quarkus/extensions/kogito-quarkus-serverless-workflow-extension/kogito-quarkus-serverless-workflow-integration-test/src/main/resources/application.properties b/quarkus/extensions/kogito-quarkus-serverless-workflow-extension/kogito-quarkus-serverless-workflow-integration-test/src/main/resources/application.properties
index bff2cf96b0a..8b83e2da2f9 100644
--- a/quarkus/extensions/kogito-quarkus-serverless-workflow-extension/kogito-quarkus-serverless-workflow-integration-test/src/main/resources/application.properties
+++ b/quarkus/extensions/kogito-quarkus-serverless-workflow-extension/kogito-quarkus-serverless-workflow-integration-test/src/main/resources/application.properties
@@ -64,6 +64,12 @@ mp.messaging.incoming.never.path=/never
 mp.messaging.incoming.eventTimeout1.connector=quarkus-http
 mp.messaging.incoming.eventTimeout1.path=/eventTimeout1
 
+mp.messaging.incoming.eventTimeout1.connector=quarkus-http
+mp.messaging.incoming.eventTimeout1.path=/eventTimeout1
+
+mp.messaging.incoming.customer-arrival-type.connector=quarkus-http
+mp.messaging.incoming.customer-arrival-type.path=/eventWithToStateFilter
+
 mp.messaging.incoming.eventTimeout2.connector=quarkus-http
 mp.messaging.incoming.eventTimeout2.path=/eventTimeout2
 
diff --git a/quarkus/extensions/kogito-quarkus-serverless-workflow-extension/kogito-quarkus-serverless-workflow-integration-test/src/main/resources/greetCustomer.sw.json b/quarkus/extensions/kogito-quarkus-serverless-workflow-extension/kogito-quarkus-serverless-workflow-integration-test/src/main/resources/greetCustomer.sw.json
new file mode 100644
index 00000000000..a3f5f1492bb
--- /dev/null
+++ b/quarkus/extensions/kogito-quarkus-serverless-workflow-extension/kogito-quarkus-serverless-workflow-integration-test/src/main/resources/greetCustomer.sw.json
@@ -0,0 +1,41 @@
+{
+"id": "GreetCustomer",
+"name": "Greet Customers when they arrive",
+"version": "1.0",
+"specVersion": "0.8",
+"start": "WaitForCustomerToArrive",
+"states":[
+{
+"name": "WaitForCustomerToArrive",
+"type": "event",
+"onEvents": [{
+"eventRefs": ["CustomerArrivesEvent"],
+"eventDataFilter": {
+"data": "${ .customer }",
+"toStateData": "${ .customerInfo }"
+},
+"actions":[
+{
+"functionRef": {
+"refName": "greetingFunction",
+"arguments": {
+"message": "${ .customerInfo.name } "
+}
+}
+}
+]
+}],
+"end": true
+}
+],
+"events": [{
+"name": "CustomerArrivesEvent",
+"type": "customer-arrival-type",
+"source": "customer-arrival-event-source"
+}],
+"functions": [{
+"name": "greetingFunction",
+ "type": "custom",
+  "operation": "sysout"
+}]
+}
\ No newline at end of file
diff --git a/quarkus/extensions/kogito-quarkus-serverless-workflow-extension/kogito-quarkus-serverless-workflow-integration-test/src/test/java/org/kie/kogito/quarkus/workflows/EventFlowIT.java b/quarkus/extensions/kogito-quarkus-serverless-workflow-extension/kogito-quarkus-serverless-workflow-integration-test/src/test/java/org/kie/kogito/quarkus/workflows/EventFlowIT.java
index 91ede4a21f9..bd96c501170 100644
--- a/quarkus/extensions/kogito-quarkus-serverless-workflow-extension/kogito-quarkus-serverless-workflow-integration-test/src/test/java/org/kie/kogito/quarkus/workflows/EventFlowIT.java
+++ b/quarkus/extensions/kogito-quarkus-serverless-workflow-extension/kogito-quarkus-serverless-workflow-integration-test/src/test/java/org/kie/kogito/quarkus/workflows/EventFlowIT.java
@@ -19,9 +19,13 @@
 package org.kie.kogito.quarkus.workflows;
 
 import java.io.IOException;
+import java.net.URI;
 import java.time.Duration;
+import java.time.OffsetDateTime;
+import java.util.Collections;
 import java.util.Map;
 import java.util.Optional;
+import java.util.UUID;
 
 import org.awaitility.core.ConditionTimeoutException;
 import org.junit.jupiter.api.BeforeAll;
@@ -32,6 +36,7 @@
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 
+import io.cloudevents.core.builder.CloudEventBuilder;
 import io.cloudevents.jackson.JsonFormat;
 import io.quarkus.test.junit.QuarkusIntegrationTest;
 import io.restassured.RestAssured;
@@ -56,6 +61,23 @@ static void init() {
         defaultMarshaller = new ByteArrayCloudEventMarshaller(mapper);
     }
 
+    @Test
+    void testStartingEventWithToStateFilter() {
+        given()
+                .contentType(ContentType.JSON)
+                .when()
+                .body(CloudEventBuilder.v1()
+                        .withId(UUID.randomUUID().toString())
+                        .withSource(URI.create("customer-arrival-event-source"))
+                        .withType("customer-arrival-type")
+                        .withTime(OffsetDateTime.now())
+                        .withData(defaultMarshaller.cloudEventDataFactory().apply(Collections.singletonMap("customer", Map.of("name", "pepe")))).build())
+                .post("/eventWithToStateFilter")
+                .then()
+                .statusCode(202);
+
+    }
+
     @Test
     void testNotStartingEvent() throws IOException {
         doIt("nonStartEvent", Optional.of("manolo"), "move");