Update listWorkflows (#448)

Add completed_after, completed_before, dequeued_after and
dequeued_before support to ListWorkflows, including conductor and admin
server.

Also fix issue where conductor doesn't return an error response on
unkown request types

fixes #447

Author: devhawk

transact/src/main/java/dev/dbos/transact/admin/ListQueuedWorkflowsRequest.java

@@ -49,7 +49,11 @@ public ListWorkflowsInput asInput() {
         parent_workflow_id,
         null, // wasForkedFrom
         null, // hasParent
-        null // attributes
+        null, // attributes
+        null, // completedAfter
+        null, // completedBefore
+        null, // dequeuedAfter
+        null // dequeuedBefore
         );
   }
 }

transact/src/main/java/dev/dbos/transact/admin/ListWorkflowsRequest.java

@@ -16,6 +16,10 @@ public record ListWorkflowsRequest(
     @JsonDeserialize(using = StringOrListDeserializer.class) List<String> authenticated_user,
     String start_time,
     String end_time,
+    String completed_after,
+    String completed_before,
+    String dequeued_after,
+    String dequeued_before,
     @JsonDeserialize(using = StringOrListDeserializer.class) List<String> status,
     @JsonDeserialize(using = StringOrListDeserializer.class) List<String> application_version,
     @JsonDeserialize(using = StringOrListDeserializer.class) List<String> fork_from,
@@ -53,7 +57,10 @@ public ListWorkflowsInput asInput() {
         parent_workflow_id,
         null, // wasForkedFrom
         null, // hasParent
-        null // attributes
-        );
+        null, // attributes
+        completed_after != null ? Instant.parse(completed_after) : null,
+        completed_before != null ? Instant.parse(completed_before) : null,
+        dequeued_after != null ? Instant.parse(dequeued_after) : null,
+        dequeued_before != null ? Instant.parse(dequeued_before) : null);
   }
 }

transact/src/main/java/dev/dbos/transact/conductor/Conductor.java

@@ -41,7 +41,11 @@ public ListWorkflowsInput toInput() {
         null, // parentWorkflowId
         null, // wasForkedFrom
         null, // hasParent
-        null // attributes
+        null, // attributes
+        null, // completedAfter
+        null, // completedBefore
+        null, // dequeuedAfter
+        null // dequeuedBefore
         );
   }
 }

transact/src/main/java/dev/dbos/transact/conductor/protocol/GetWorkflowRequest.java

@@ -87,6 +87,11 @@ public ListWorkflowsInput asInput() {
         body.parent_workflow_id,
         body.was_forked_from,
         body.has_parent,
-        body.attributes);
+        body.attributes,
+        null, // completedAfter
+        null, // completedBefore
+        null, // dequeuedAfter
+        null // dequeuedBefore
+        );
   }
 }

transact/src/main/java/dev/dbos/transact/conductor/protocol/ListQueuedWorkflowsRequest.java

@@ -26,6 +26,10 @@ public static class Body {
 
     public String start_time;
     public String end_time;
+    public String completed_after;
+    public String completed_before;
+    public String dequeued_after;
+    public String dequeued_before;
 
     @JsonDeserialize(using = StringOrListDeserializer.class)
     public List<String> status;
@@ -85,6 +89,10 @@ public ListWorkflowsInput asInput() {
         body.parent_workflow_id,
         body.was_forked_from,
         body.has_parent,
-        body.attributes);
+        body.attributes,
+        body.completed_after != null ? Instant.parse(body.completed_after) : null,
+        body.completed_before != null ? Instant.parse(body.completed_before) : null,
+        body.dequeued_after != null ? Instant.parse(body.dequeued_after) : null,
+        body.dequeued_before != null ? Instant.parse(body.dequeued_before) : null);
   }
 }

transact/src/main/java/dev/dbos/transact/conductor/protocol/ListWorkflowsRequest.java

@@ -52,7 +52,7 @@ public static MessageType fromValue(String value) {
         return type;
       }
     }
-    throw new IllegalArgumentException("Unknown message type: " + value);
+    return null;
   }
 
   @Override

transact/src/main/java/dev/dbos/transact/conductor/protocol/MessageType.java

@@ -728,6 +728,22 @@ public static List<WorkflowStatus> listWorkflows(DbContext ctx, ListWorkflowsInp
       whereConditions.add("created_at <= ?");
       parameters.add(input.endTime().toEpochMilli());
     }
+    if (input.completedAfter() != null) {
+      whereConditions.add("completed_at >= ?");
+      parameters.add(input.completedAfter().toEpochMilli());
+    }
+    if (input.completedBefore() != null) {
+      whereConditions.add("completed_at <= ?");
+      parameters.add(input.completedBefore().toEpochMilli());
+    }
+    if (input.dequeuedAfter() != null) {
+      whereConditions.add("started_at_epoch_ms >= ?");
+      parameters.add(input.dequeuedAfter().toEpochMilli());
+    }
+    if (input.dequeuedBefore() != null) {
+      whereConditions.add("started_at_epoch_ms <= ?");
+      parameters.add(input.dequeuedBefore().toEpochMilli());
+    }
     if (input.status() != null && !input.status().isEmpty()) {
       whereConditions.add("status = ANY(?)");
       parameters.add(input.status());