Get Workflow Aggregates, Events, Notifications & Streams (#344)

Author: devhawk

transact/src/main/java/dev/dbos/transact/DBOS.java

@@ -37,6 +37,7 @@
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Properties;
@@ -943,6 +944,19 @@ public void applySchedules(@NonNull List<WorkflowSchedule> schedules) {
     return ensureLaunched("retrieveWorkflow").retrieveWorkflow(workflowId);
   }
 
+  /**
+   * Retrieves all events stored during the execution of a workflow. Events are key-value pairs that
+   * workflows can set during execution to persist intermediate state or communicate between steps.
+   * This method returns all events for the specified workflow with their deserialized values.
+   *
+   * @param workflowId the unique identifier of the workflow whose events to retrieve
+   * @return a map containing all events for the workflow, where keys are event names and values are
+   *     the deserialized event data
+   */
+  public @NonNull Map<String, Object> getAllEvents(@NonNull String workflowId) {
+    return ensureLaunched("getAllEvents").getAllEvents(workflowId);
+  }
+
   /**
    * List all workflows
    *

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

@@ -684,6 +684,10 @@ CompletableFuture<BaseResponse> getResponseAsync(BaseMessage message, WebSocket
       case FORK_WORKFLOW -> handleFork(this, message);
       case GET_METRICS -> handleGetMetrics(this, message);
       case GET_SCHEDULE -> handleGetSchedule(this, message);
+      case GET_WORKFLOW_AGGREGATES -> handleGetWorkflowAggregates(this, message);
+      case GET_WORKFLOW_EVENTS -> handleGetWorkflowEvents(this, message);
+      case GET_WORKFLOW_NOTIFICATIONS -> handleGetWorkflowNotifications(this, message);
+      case GET_WORKFLOW_STREAMS -> handleGetWorkflowStreams(this, message);
       case GET_WORKFLOW -> handleGetWorkflow(this, message);
       case IMPORT_WORKFLOW -> handleImportWorkflow(this, message);
       case LIST_APPLICATION_VERSIONS -> handleListApplicationVersions(this, message);
@@ -988,6 +992,75 @@ static CompletableFuture<BaseResponse> handleGetMetrics(
         });
   }
 
+  static CompletableFuture<BaseResponse> handleGetWorkflowAggregates(
+      Conductor conductor, BaseMessage message) {
+    return CompletableFuture.supplyAsync(
+        () -> {
+          GetWorkflowAggregatesRequest request = (GetWorkflowAggregatesRequest) message;
+          try {
+            var rows = conductor.systemDatabase.getWorkflowAggregates(request.toInput());
+            return new GetWorkflowAggregatesResponse(request, rows);
+          } catch (Exception e) {
+            logger.error("Exception encountered when getting workflow aggregates", e);
+            return new GetWorkflowAggregatesResponse(request, e);
+          }
+        });
+  }
+
+  static CompletableFuture<BaseResponse> handleGetWorkflowEvents(
+      Conductor conductor, BaseMessage message) {
+    return CompletableFuture.supplyAsync(
+        () -> {
+          GetWorkflowEventsRequest request = (GetWorkflowEventsRequest) message;
+          try {
+            var events = conductor.systemDatabase.getAllEvents(request.workflow_id);
+            return new GetWorkflowEventsResponse(request, events);
+          } catch (Exception e) {
+            logger.error(
+                "Exception encountered when getting workflow events for {}",
+                request.workflow_id,
+                e);
+            return new GetWorkflowEventsResponse(request, e);
+          }
+        });
+  }
+
+  static CompletableFuture<BaseResponse> handleGetWorkflowNotifications(
+      Conductor conductor, BaseMessage message) {
+    return CompletableFuture.supplyAsync(
+        () -> {
+          GetWorkflowNotificationsRequest request = (GetWorkflowNotificationsRequest) message;
+          try {
+            var notifications = conductor.systemDatabase.getAllNotifications(request.workflow_id);
+            return new GetWorkflowNotificationsResponse(request, notifications);
+          } catch (Exception e) {
+            logger.error(
+                "Exception encountered when getting workflow notifications for {}",
+                request.workflow_id,
+                e);
+            return new GetWorkflowNotificationsResponse(request, e);
+          }
+        });
+  }
+
+  static CompletableFuture<BaseResponse> handleGetWorkflowStreams(
+      Conductor conductor, BaseMessage message) {
+    return CompletableFuture.supplyAsync(
+        () -> {
+          GetWorkflowStreamsRequest request = (GetWorkflowStreamsRequest) message;
+          try {
+            var streams = conductor.systemDatabase.getAllStreamEntries(request.workflow_id);
+            return new GetWorkflowStreamsResponse(request, streams);
+          } catch (Exception e) {
+            logger.error(
+                "Exception encountered when getting workflow streams for {}",
+                request.workflow_id,
+                e);
+            return new GetWorkflowStreamsResponse(request, e);
+          }
+        });
+  }
+
   private static boolean isImportMessage(CharSequence data) {
     int checkLen = Math.min(data.length(), 200);
     return data.subSequence(0, checkLen).toString().contains("\"import_workflow\"");

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

@@ -20,6 +20,12 @@
   @JsonSubTypes.Type(value = ForkWorkflowRequest.class, name = "fork_workflow"),
   @JsonSubTypes.Type(value = GetMetricsRequest.class, name = "get_metrics"),
   @JsonSubTypes.Type(value = GetScheduleRequest.class, name = "get_schedule"),
+  @JsonSubTypes.Type(value = GetWorkflowAggregatesRequest.class, name = "get_workflow_aggregates"),
+  @JsonSubTypes.Type(value = GetWorkflowEventsRequest.class, name = "get_workflow_events"),
+  @JsonSubTypes.Type(
+      value = GetWorkflowNotificationsRequest.class,
+      name = "get_workflow_notifications"),
+  @JsonSubTypes.Type(value = GetWorkflowStreamsRequest.class, name = "get_workflow_streams"),
   @JsonSubTypes.Type(value = GetWorkflowRequest.class, name = "get_workflow"),
   @JsonSubTypes.Type(value = ImportWorkflowRequest.class, name = "import_workflow"),
   @JsonSubTypes.Type(

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

@@ -0,0 +1,51 @@
+package dev.dbos.transact.conductor.protocol;
+
+import dev.dbos.transact.workflow.GetWorkflowAggregatesInput;
+
+import java.time.Instant;
+import java.util.List;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class GetWorkflowAggregatesRequest extends BaseMessage {
+
+  public Body body;
+
+  @JsonIgnoreProperties(ignoreUnknown = true)
+  public static class Body {
+    public boolean group_by_status;
+    public boolean group_by_name;
+    public boolean group_by_queue_name;
+    public boolean group_by_executor_id;
+    public boolean group_by_application_version;
+    public List<String> status;
+    public String start_time;
+    public String end_time;
+    public List<String> name;
+    public List<String> app_version;
+    public List<String> executor_id;
+    public List<String> queue_name;
+    public List<String> workflow_id_prefix;
+  }
+
+  public GetWorkflowAggregatesInput toInput() {
+    if (body == null) {
+      return new GetWorkflowAggregatesInput();
+    }
+    return new GetWorkflowAggregatesInput(
+        body.group_by_status,
+        body.group_by_name,
+        body.group_by_queue_name,
+        body.group_by_executor_id,
+        body.group_by_application_version,
+        body.name,
+        body.status,
+        body.queue_name,
+        body.executor_id,
+        body.app_version,
+        body.workflow_id_prefix,
+        body.start_time != null ? Instant.parse(body.start_time) : null,
+        body.end_time != null ? Instant.parse(body.end_time) : null);
+  }
+}

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

@@ -0,0 +1,30 @@
+package dev.dbos.transact.conductor.protocol;
+
+import dev.dbos.transact.workflow.WorkflowAggregateRow;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+public class GetWorkflowAggregatesResponse extends BaseResponse {
+
+  public static record WorkflowAggregateOutput(Map<String, String> group, long count) {
+    public static WorkflowAggregateOutput from(WorkflowAggregateRow row) {
+      return new WorkflowAggregateOutput(row.group(), row.count());
+    }
+  }
+
+  public List<WorkflowAggregateOutput> output;
+
+  public GetWorkflowAggregatesResponse() {}
+
+  public GetWorkflowAggregatesResponse(BaseMessage message, List<WorkflowAggregateRow> rows) {
+    super(message.type, message.request_id);
+    this.output = rows.stream().map(WorkflowAggregateOutput::from).toList();
+  }
+
+  public GetWorkflowAggregatesResponse(BaseMessage message, Exception ex) {
+    super(message.type, message.request_id, ex.getMessage());
+    this.output = Collections.emptyList();
+  }
+}

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

@@ -0,0 +1,8 @@
+package dev.dbos.transact.conductor.protocol;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class GetWorkflowEventsRequest extends BaseMessage {
+  public String workflow_id;
+}

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

@@ -0,0 +1,30 @@
+package dev.dbos.transact.conductor.protocol;
+
+import dev.dbos.transact.json.JSONUtil;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+public class GetWorkflowEventsResponse extends BaseResponse {
+
+  public record EventOutput(String key, String value) {
+    public static EventOutput from(Map.Entry<String, Object> entry) {
+      return new EventOutput(entry.getKey(), JSONUtil.toJson(entry.getValue()));
+    }
+  }
+
+  public List<EventOutput> events;
+
+  public GetWorkflowEventsResponse() {}
+
+  public GetWorkflowEventsResponse(BaseMessage message, Map<String, Object> events) {
+    super(message.type, message.request_id);
+    this.events = events.entrySet().stream().map(EventOutput::from).toList();
+  }
+
+  public GetWorkflowEventsResponse(BaseMessage message, Exception ex) {
+    super(message.type, message.request_id, ex.getMessage());
+    this.events = Collections.emptyList();
+  }
+}

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

@@ -0,0 +1,8 @@
+package dev.dbos.transact.conductor.protocol;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class GetWorkflowNotificationsRequest extends BaseMessage {
+  public String workflow_id;
+}

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

@@ -0,0 +1,32 @@
+package dev.dbos.transact.conductor.protocol;
+
+import dev.dbos.transact.json.JSONUtil;
+import dev.dbos.transact.workflow.NotificationInfo;
+
+import java.util.Collections;
+import java.util.List;
+
+public class GetWorkflowNotificationsResponse extends BaseResponse {
+
+  public record NotificationOutput(
+      String topic, String message, long created_at_epoch_ms, boolean consumed) {
+    public static NotificationOutput from(NotificationInfo info) {
+      return new NotificationOutput(
+          info.topic(), JSONUtil.toJson(info.message()), info.createdAtEpochMs(), info.consumed());
+    }
+  }
+
+  public List<NotificationOutput> notifications;
+
+  public GetWorkflowNotificationsResponse() {}
+
+  public GetWorkflowNotificationsResponse(BaseMessage message, List<NotificationInfo> infos) {
+    super(message.type, message.request_id);
+    this.notifications = infos.stream().map(NotificationOutput::from).toList();
+  }
+
+  public GetWorkflowNotificationsResponse(BaseMessage message, Exception ex) {
+    super(message.type, message.request_id, ex.getMessage());
+    this.notifications = Collections.emptyList();
+  }
+}

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

@@ -0,0 +1,8 @@
+package dev.dbos.transact.conductor.protocol;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class GetWorkflowStreamsRequest extends BaseMessage {
+  public String workflow_id;
+}