Search workflow instances by state variable key and value

CHANGELOG.md

@@ -2,12 +2,15 @@
 
 **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

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

@@ -628,46 +628,56 @@ 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());
     if (!isEmpty(query.ids)) {
-      conditions.add("id in (:ids)");
+      conditions.add("wf.id in (:ids)");
       params.addValue("ids", query.ids);
     }
     if (!isEmpty(query.types)) {
-      conditions.add("type in (:types)");
+      conditions.add("wf.type in (:types)");
       params.addValue("types", query.types);
     }
     if (query.parentWorkflowId != null) {
-      conditions.add("parent_workflow_id = :parent_workflow_id");
+      conditions.add("wf.parent_workflow_id = :parent_workflow_id");
       params.addValue("parent_workflow_id", query.parentWorkflowId);
     }
     if (query.parentActionId != null) {
-      conditions.add("parent_action_id = :parent_action_id");
+      conditions.add("wf.parent_action_id = :parent_action_id");
       params.addValue("parent_action_id", query.parentActionId);
     }
     if (!isEmpty(query.states)) {
-      conditions.add("state in (:states)");
+      conditions.add("wf.state in (:states)");
       params.addValue("states", query.states);
     }
     if (!isEmpty(query.statuses)) {
       List<String> convertedStatuses = query.statuses.stream().map(WorkflowInstanceStatus::name).collect(toList());
-      conditions.add("status" + sqlVariants.castToText() + " in (:statuses)");
+      conditions.add("wf.status" + sqlVariants.castToText() + " in (:statuses)");
       params.addValue("statuses", convertedStatuses);
     }
     if (query.businessKey != null) {
-      conditions.add("business_key = :business_key");
+      conditions.add("wf.business_key = :business_key");
       params.addValue("business_key", query.businessKey);
     }
     if (query.externalId != null) {
-      conditions.add("external_id = :external_id");
+      conditions.add("wf.external_id = :external_id");
       params.addValue("external_id", query.externalId);
     }
-    conditions.add("executor_group = :executor_group");
+    conditions.add("wf.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 outside on wf.id = outside.workflow_id ")
+          .append("inner join (select workflow_id, max(action_id) action_id from nflow_workflow_state ")
+          .append("where state_key = :state_key group by workflow_id) inside ")
+          .append("on outside.workflow_id = inside.workflow_id and outside.action_id = inside.action_id ");
+      conditions.add("outside.state_key = :state_key");
+      params.addValue("state_key", query.stateVariableKey);
+      conditions.add("outside.state_value = :state_value");
+      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-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)