Search workflow instances by state variable key and value (#421)

* Add support for searching worklflow instances by state variable key and value

* Add support for searching workflow instances by state variable key and value via REST API

* refactor sql for state var query and add test

* Change  data types to  for MS SQL Server

Co-authored-by: Edvard Fonsell <edvard.fonsell@nitorcreations.com>

CHANGELOG.md

@@ -2,16 +2,21 @@
 
 **Highlights**
 - Support updating workflow instance business key.
+- Support for searching workflow instances by state variable key and value.
 
 **Details**
 - `nflow-engine`
   - `WorkflowInstanceService.updateWorkflowInstance` can now be used to update business key of the workflow instance.
-- `nflow-rest-api-common`
+  - Use `QueryWorkflowInstances.setStateVariable` to limit search query by state variable name and key. Only the latest value of the state variable of the workflow instance is used.
+- `nflow-rest-api-common`, `nflow-rest-api-jax-rs`, `nflow-rest-api-spring-web`
   - `UpdateWorkflowInstanceRequest.businessKey` field was added to support updating workflow instance business key via REST API.
+  - Added support for new query parameters `stateVariableKey` and `stateVariableValue` to `GET /v1/workflow-instance` to limit search query by state variable name and key. Only the latest value of the state variable of the workflow instance is used.
 - `nflow-explorer`
   - Dependency updates:
     - urijs 1.19.5
     - socket.io 2.4.1
+- Database
+  - Change `text` data types to `varchar(max)` for MS SQL Server
 
 ## 7.2.2 (2020-12-25)
 

nflow-engine/src/main/java/io/nflow/engine/internal/dao/WorkflowInstanceDao.java

@@ -628,7 +628,7 @@ public List<WorkflowInstance> queryWorkflowInstances(QueryWorkflowInstances quer
   }
 
   public Stream<WorkflowInstance> queryWorkflowInstancesAsStream(QueryWorkflowInstances query) {
-    String sql = "select * from nflow_workflow ";
+    StringBuilder sqlBuilder = new StringBuilder("select wf.* from nflow_workflow wf ");
     List<String> conditions = new ArrayList<>();
     MapSqlParameterSource params = new MapSqlParameterSource();
     conditions.add(executorInfo.getExecutorGroupCondition());
@@ -667,7 +667,13 @@ public Stream<WorkflowInstance> queryWorkflowInstancesAsStream(QueryWorkflowInst
     }
     conditions.add("executor_group = :executor_group");
     params.addValue("executor_group", executorInfo.getExecutorGroup());
-    sql += " where " + collectionToDelimitedString(conditions, " and ") + " order by id desc";
+    if (query.stateVariableKey != null) {
+      sqlBuilder.append("inner join nflow_workflow_state wfs on wf.id = wfs.workflow_id and wfs.state_key = :state_key and wfs.state_value = :state_value ");
+      conditions.add("wfs.action_id = (select max(action_id) from nflow_workflow_state where workflow_id = wf.id and state_key = :state_key)");
+      params.addValue("state_key", query.stateVariableKey);
+      params.addValue("state_value", query.stateVariableValue);
+    }
+    String sql = sqlBuilder.append("where ").append(collectionToDelimitedString(conditions, " and ")).append(" order by id desc").toString();
     sql = sqlVariants.limit(sql, getMaxResults(query.maxResults));
 
     Stream<WorkflowInstance> ret = namedJdbc.query(sql, params, new WorkflowInstanceRowMapper()).stream()

nflow-engine/src/main/java/io/nflow/engine/workflow/instance/QueryWorkflowInstances.java

@@ -5,6 +5,8 @@
 import java.util.ArrayList;
 import java.util.List;
 
+import org.springframework.util.Assert;
+
 import io.nflow.engine.model.ModelObject;
 import io.nflow.engine.workflow.instance.WorkflowInstance.WorkflowInstanceStatus;
 
@@ -53,6 +55,16 @@ public class QueryWorkflowInstances extends ModelObject {
    */
   public final String externalId;
 
+  /**
+   * State variable key.
+   */
+  public final String stateVariableKey;
+
+  /**
+   * State variable value.
+   */
+  public final String stateVariableValue;
+
   /**
    * Setting this to true will make the query return also workflow actions.
    */
@@ -94,6 +106,8 @@ public class QueryWorkflowInstances extends ModelObject {
     this.statuses = new ArrayList<>(builder.statuses);
     this.businessKey = builder.businessKey;
     this.externalId = builder.externalId;
+    this.stateVariableKey = builder.stateVariableKey;
+    this.stateVariableValue = builder.stateVariableValue;
     this.includeActions = builder.includeActions;
     this.includeCurrentStateVariables = builder.includeCurrentStateVariables;
     this.includeActionStateVariables = builder.includeActionStateVariables;
@@ -114,6 +128,8 @@ public static class Builder {
     List<WorkflowInstanceStatus> statuses = new ArrayList<>();
     String businessKey;
     String externalId;
+    String stateVariableKey;
+    String stateVariableValue;
     boolean includeActions;
     boolean includeCurrentStateVariables;
     boolean includeActionStateVariables;
@@ -136,6 +152,8 @@ public Builder(QueryWorkflowInstances copy) {
       this.statuses = copy.statuses;
       this.businessKey = copy.businessKey;
       this.externalId = copy.externalId;
+      this.stateVariableKey = copy.stateVariableKey;
+      this.stateVariableValue = copy.stateVariableValue;
       this.includeActions = copy.includeActions;
       this.includeCurrentStateVariables = copy.includeCurrentStateVariables;
       this.includeActionStateVariables = copy.includeActionStateVariables;
@@ -223,6 +241,20 @@ public Builder setExternalId(String externalId) {
       return this;
     }
 
+    /**
+     * Set state variable key and value to query parameters.
+     * @param stateVariableKey State variable key.
+     * @param stateVariableValue State variable vaue.
+     * @return this.
+     */
+    public Builder setStateVariable(String stateVariableKey, String stateVariableValue) {
+      Assert.notNull("stateVariableKey cannot be null", stateVariableKey);
+      Assert.notNull("stateVariableValue cannot be null", stateVariableValue);
+      this.stateVariableKey = stateVariableKey;
+      this.stateVariableValue = stateVariableValue;
+      return this;
+    }
+
     /**
      * Set whether workflow actions should be included in the results. Default is `false`.
      * @param includeActions True to include actions, false otherwise.

nflow-engine/src/main/resources/scripts/db/sqlserver.create.ddl.sql

@@ -60,7 +60,7 @@ create table nflow_workflow_state (
   workflow_id int not null,
   action_id int not null,
   state_key varchar(64) not null,
-  state_value text not null,
+  state_value varchar(max) not null,
   constraint pk_workflow_state primary key (workflow_id, action_id, state_key),
   constraint fk_state_workflow_id foreign key (workflow_id) references nflow_workflow(id)
 );
@@ -84,7 +84,7 @@ if not exists (select 1 from sys.tables where name='nflow_workflow_definition')
 create table nflow_workflow_definition (
   type varchar(64) not null,
   definition_sha1 varchar(40) not null,
-  definition text not null,
+  definition varchar(max) not null,
   created datetimeoffset(3) not null default SYSDATETIMEOFFSET(),
   modified datetimeoffset(3) not null default SYSDATETIMEOFFSET(),
   modified_by int not null,
@@ -155,7 +155,7 @@ create table nflow_archive_workflow_state (
   workflow_id int not null,
   action_id int not null,
   state_key varchar(64) not null,
-  state_value text not null,
+  state_value varchar(max) not null,
   constraint pk_arch_workflow_state primary key (workflow_id, action_id, state_key),
   constraint fk_arch_state_wf_id foreign key (workflow_id) references nflow_archive_workflow(id)
 );

nflow-engine/src/main/resources/scripts/db/update-7.2.2-x/sqlserver.update.ddl.sql

@@ -0,0 +1,5 @@
+alter table nflow_workflow_state modify column state_value varchar(max) not null;
+
+alter table nflow_workflow_definition modify column definition varchar(max) not null;
+
+alter table nflow_workflow_state modify column state_value varchar(max) not null;

nflow-rest-api-common/src/main/java/io/nflow/rest/v1/ResourceBase.java

@@ -139,27 +139,28 @@ public boolean updateWorkflowInstance(final long id,
     return workflowInstances.updateWorkflowInstance(instance, action);
   }
 
-  public Stream<ListWorkflowInstanceResponse> listWorkflowInstances(final List<Long> ids, final List<String> types,
-      final Long parentWorkflowId, final Long parentActionId, final List<String> states,
-      final List<WorkflowInstanceStatus> statuses, final String businessKey, final String externalId, final String include,
-      final Long maxResults, final Long maxActions, final WorkflowInstanceService workflowInstances,
-      final ListWorkflowInstanceConverter listWorkflowConverter) {
+  public Stream<ListWorkflowInstanceResponse> listWorkflowInstances(List<Long> ids, List<String> types, Long parentWorkflowId,
+      Long parentActionId, List<String> states, List<WorkflowInstanceStatus> statuses, String businessKey, String externalId,
+      String stateVariableKey, String stateVariableValue, String include, Long maxResults, Long maxActions,
+      WorkflowInstanceService workflowInstances, ListWorkflowInstanceConverter listWorkflowConverter) {
     Set<String> includeStrings = parseIncludeStrings(include).collect(toSet());
-    QueryWorkflowInstances q = new QueryWorkflowInstances.Builder() //
-        .addIds(ids.toArray(new Long[ids.size()])) //
-        .addTypes(types.toArray(new String[types.size()])) //
-        .setParentWorkflowId(parentWorkflowId) //
-        .setParentActionId(parentActionId) //
-        .addStates(states.toArray(new String[states.size()])) //
-        .addStatuses(statuses.toArray(new WorkflowInstanceStatus[statuses.size()])) //
-        .setBusinessKey(businessKey) //
-        .setExternalId(externalId) //
-        .setIncludeCurrentStateVariables(includeStrings.contains(currentStateVariables)) //
-        .setIncludeActions(includeStrings.contains(actions)) //
-        .setIncludeActionStateVariables(includeStrings.contains(actionStateVariables)) //
-        .setMaxResults(maxResults) //
-        .setMaxActions(maxActions) //
-        .setIncludeChildWorkflows(includeStrings.contains(childWorkflows)).build();
+    QueryWorkflowInstances q = new QueryWorkflowInstances.Builder()
+        .addIds(ids.toArray(new Long[ids.size()]))
+        .addTypes(types.toArray(new String[types.size()]))
+        .setParentWorkflowId(parentWorkflowId)
+        .setParentActionId(parentActionId)
+        .addStates(states.toArray(new String[states.size()]))
+        .addStatuses(statuses.toArray(new WorkflowInstanceStatus[statuses.size()]))
+        .setBusinessKey(businessKey)
+        .setExternalId(externalId)
+        .setIncludeCurrentStateVariables(includeStrings.contains(currentStateVariables))
+        .setIncludeActions(includeStrings.contains(actions))
+        .setIncludeActionStateVariables(includeStrings.contains(actionStateVariables))
+        .setMaxResults(maxResults)
+        .setMaxActions(maxActions)
+        .setIncludeChildWorkflows(includeStrings.contains(childWorkflows))
+        .setStateVariable(stateVariableKey, stateVariableValue)
+        .build();
     Stream<WorkflowInstance> instances = workflowInstances.listWorkflowInstancesAsStream(q);
     Set<WorkflowInstanceInclude> parseIncludeEnums = parseIncludeEnums(include);
     return instances.map(instance -> listWorkflowConverter.convert(instance, parseIncludeEnums));

nflow-rest-api-jax-rs/src/main/java/io/nflow/rest/v1/jaxrs/WorkflowInstanceResource.java

@@ -137,11 +137,14 @@ public Response listWorkflowInstances(@QueryParam("id") @ApiParam("Internal id o
       @QueryParam("status") @ApiParam("Current status of workflow instance") List<WorkflowInstanceStatus> statuses,
       @QueryParam("businessKey") @ApiParam("Business key for workflow instance") String businessKey,
       @QueryParam("externalId") @ApiParam("External id for workflow instance") String externalId,
+      @QueryParam("stateVariableKey") @ApiParam("Key of state variable that must exist for workflow instance") String stateVariableKey,
+      @QueryParam("stateVariableValue") @ApiParam("Current value of state variable defined by stateVariableKey") String stateVariableValue,
       @QueryParam("include") @ApiParam(value = INCLUDE_PARAM_DESC, allowableValues = INCLUDE_PARAM_VALUES, allowMultiple = true) String include,
       @QueryParam("maxResults") @ApiParam("Maximum number of workflow instances to be returned") Long maxResults,
       @QueryParam("maxActions") @ApiParam("Maximum number of actions returned for each workflow instance") Long maxActions) {
     return handleExceptions(() -> ok(super.listWorkflowInstances(ids, types, parentWorkflowId, parentActionId, states, statuses,
-        businessKey, externalId, include, maxResults, maxActions, workflowInstances, listWorkflowConverter).iterator()));
+        businessKey, externalId, stateVariableKey, stateVariableValue, include, maxResults, maxActions, workflowInstances,
+        listWorkflowConverter).iterator()));
   }
 
   @PUT

nflow-rest-api-jax-rs/src/test/java/io/nflow/rest/v1/jaxrs/WorkflowInstanceResourceTest.java

@@ -11,6 +11,7 @@
 import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.nullValue;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -212,29 +213,49 @@ public void whenUpdatingBusinessKeyWithDescriptionUpdateWorkflowInstanceWorks()
   @Test
   public void listWorkflowInstancesWorks() {
     makeRequest(() -> resource.listWorkflowInstances(asList(42L), asList("type"), 99L, 88L, asList("state"),
-        asList(WorkflowInstanceStatus.created), "businessKey", "externalId", "", null, null));
-    verify(workflowInstances).listWorkflowInstancesAsStream((QueryWorkflowInstances) argThat(allOf(hasField("ids", contains(42L)),
-        hasField("types", contains("type")), hasField("parentWorkflowId", is(99L)), hasField("parentActionId", is(88L)),
-        hasField("states", contains("state")), hasField("statuses", contains(WorkflowInstanceStatus.created)),
-        hasField("businessKey", equalTo("businessKey")), hasField("externalId", equalTo("externalId")),
-        hasField("includeActions", equalTo(false)), hasField("includeCurrentStateVariables", equalTo(false)),
-        hasField("includeActionStateVariables", equalTo(false)), hasField("includeChildWorkflows", equalTo(false)),
-        hasField("maxResults", equalTo(null)), hasField("maxActions", equalTo(null)))));
+        asList(WorkflowInstanceStatus.created), "businessKey", "externalId", null, null, "", null, null));
+    verify(workflowInstances).listWorkflowInstancesAsStream((QueryWorkflowInstances) argThat(allOf(
+        hasField("ids", contains(42L)),
+        hasField("types", contains("type")),
+        hasField("parentWorkflowId", is(99L)),
+        hasField("parentActionId", is(88L)),
+        hasField("states", contains("state")),
+        hasField("statuses", contains(WorkflowInstanceStatus.created)),
+        hasField("businessKey", equalTo("businessKey")),
+        hasField("externalId", equalTo("externalId")),
+        hasField("stateVariableKey", nullValue()),
+        hasField("stateVariableValue", nullValue()),
+        hasField("includeActions", equalTo(false)),
+        hasField("includeCurrentStateVariables", equalTo(false)),
+        hasField("includeActionStateVariables", equalTo(false)),
+        hasField("includeChildWorkflows", equalTo(false)),
+        hasField("maxResults", equalTo(null)),
+        hasField("maxActions", equalTo(null)))));
   }
 
   @Test
   public void listWorkflowInstancesWorksWithAllIncludes() {
     makeRequest(() -> resource.listWorkflowInstances(asList(42L), asList("type"), 99L, 88L, asList("state"),
         asList(WorkflowInstanceStatus.created, WorkflowInstanceStatus.executing), "businessKey", "externalId",
-        "actions,currentStateVariables,actionStateVariables,childWorkflows", 1L, 1L));
+        "stateVarKey", "stateVarValue", "actions,currentStateVariables,actionStateVariables,childWorkflows", 1L, 1L));
     verify(workflowInstances).listWorkflowInstancesAsStream(
-        (QueryWorkflowInstances) argThat(allOf(hasField("ids", contains(42L)), hasField("types", contains("type")),
-            hasField("parentWorkflowId", is(99L)), hasField("parentActionId", is(88L)), hasField("states", contains("state")),
+        (QueryWorkflowInstances) argThat(allOf(
+            hasField("ids", contains(42L)),
+            hasField("types", contains("type")),
+            hasField("parentWorkflowId", is(99L)),
+            hasField("parentActionId", is(88L)),
+            hasField("states", contains("state")),
             hasField("statuses", contains(WorkflowInstanceStatus.created, WorkflowInstanceStatus.executing)),
-            hasField("businessKey", equalTo("businessKey")), hasField("externalId", equalTo("externalId")),
-            hasField("includeActions", equalTo(true)), hasField("includeCurrentStateVariables", equalTo(true)),
-            hasField("includeActionStateVariables", equalTo(true)), hasField("includeChildWorkflows", equalTo(true)),
-            hasField("maxResults", equalTo(1L)), hasField("maxActions", equalTo(1L)))));
+            hasField("businessKey", equalTo("businessKey")),
+            hasField("externalId", equalTo("externalId")),
+            hasField("stateVariableKey", equalTo("stateVarKey")),
+            hasField("stateVariableValue", equalTo("stateVarValue")),
+            hasField("includeActions", equalTo(true)),
+            hasField("includeCurrentStateVariables", equalTo(true)),
+            hasField("includeActionStateVariables", equalTo(true)),
+            hasField("includeChildWorkflows", equalTo(true)),
+            hasField("maxResults", equalTo(1L)),
+            hasField("maxActions", equalTo(1L)))));
   }
 
   @Test

nflow-rest-api-spring-web/src/main/java/io/nflow/rest/v1/springweb/WorkflowInstanceResource.java

@@ -130,12 +130,15 @@ public Mono<ResponseEntity<?>> listWorkflowInstances(
       @RequestParam(value = "status", defaultValue = "") @ApiParam("Current status of workflow instance") List<WorkflowInstanceStatus> statuses,
       @RequestParam(value = "businessKey", required = false) @ApiParam("Business key for workflow instance") String businessKey,
       @RequestParam(value = "externalId", required = false) @ApiParam("External id for workflow instance") String externalId,
+      @RequestParam(value = "stateVariableKey", required = false) @ApiParam("Key of state variable that must exist for workflow instance") String stateVariableKey,
+      @RequestParam(value = "stateVariableValue", required = false) @ApiParam("Current value of state variable defined by stateVariableKey") String stateVariableValue,
       @RequestParam(value = "include", required = false) @ApiParam(value = INCLUDE_PARAM_DESC, allowableValues = INCLUDE_PARAM_VALUES, allowMultiple = true) String include,
       @RequestParam(value = "maxResults", required = false) @ApiParam("Maximum number of workflow instances to be returned") Long maxResults,
       @RequestParam(value = "maxActions", required = false) @ApiParam("Maximum number of actions returned for each workflow instance") Long maxActions) {
     return handleExceptions(() -> wrapBlocking(
         () -> ok(super.listWorkflowInstances(ids, types, parentWorkflowId, parentActionId, states, statuses, businessKey,
-            externalId, include, maxResults, maxActions, this.workflowInstances, this.listWorkflowConverter).iterator())));
+            externalId, stateVariableKey, stateVariableValue, include, maxResults, maxActions, this.workflowInstances,
+            this.listWorkflowConverter).iterator())));
   }
 
   @PutMapping(path = "/{id}/signal", consumes = APPLICATION_JSON_VALUE)

nflow-tests/src/main/java/io/nflow/tests/demo/workflow/StateWorkflow.java

@@ -28,6 +28,7 @@
 public class StateWorkflow extends WorkflowDefinition<StateWorkflow.State> {
 
   public static final String STATE_WORKFLOW_TYPE = "stateWorkflow";
+  public static final String STATEVAR_QUERYTEST = "queryTest";
 
   public static enum State implements io.nflow.engine.workflow.definition.WorkflowState {
     state1(start, "Set variable 1"),
@@ -71,18 +72,21 @@ public StateWorkflow() {
   public NextAction state1(@SuppressWarnings("unused") StateExecution execution,
       @StateVar(value = "variable1", instantiateIfNotExists = true) Variable variable1) {
     variable1.value = "foo1";
+    execution.setVariable(STATEVAR_QUERYTEST, "oldValue");
     return moveToState(state2, "variable1 is set to " + variable1.value);
   }
 
   public NextAction state2(@SuppressWarnings("unused") StateExecution execution,
       @StateVar(value = "variable2", instantiateIfNotExists = true) Variable variable2) {
     variable2.value = "bar1";
+    execution.setVariable(STATEVAR_QUERYTEST, "anotherOldValue");
     return moveToState(state3, "variable1 is set to " + variable2.value);
   }
 
   public NextAction state3(@SuppressWarnings("unused") StateExecution execution,
       @StateVar(value = "variable2") Variable variable2) {
     variable2.value = "bar2";
+    execution.setVariable(STATEVAR_QUERYTEST, "newValue");
     return moveToState(state4, "variable2 is set to " + variable2.value);
   }