DBOSClient.java

package dev.dbos.transact;

import static dev.dbos.transact.internal.Validation.nullableIsEmpty;
import static dev.dbos.transact.internal.Validation.nullableIsNotPositive;

import dev.dbos.transact.database.Result;
import dev.dbos.transact.database.StreamIterator;
import dev.dbos.transact.database.SystemDatabase;
import dev.dbos.transact.execution.DBOSExecutor;
import dev.dbos.transact.execution.ExecutionOptions;
import dev.dbos.transact.json.DBOSSerializer;
import dev.dbos.transact.json.PortableWorkflowException;
import dev.dbos.transact.json.SerializationUtil;
import dev.dbos.transact.workflow.ForkOptions;
import dev.dbos.transact.workflow.ListWorkflowsInput;
import dev.dbos.transact.workflow.Queue;
import dev.dbos.transact.workflow.QueueConflictResolution;
import dev.dbos.transact.workflow.QueueOptions;
import dev.dbos.transact.workflow.ScheduleStatus;
import dev.dbos.transact.workflow.SendMessage;
import dev.dbos.transact.workflow.SerializationStrategy;
import dev.dbos.transact.workflow.StepInfo;
import dev.dbos.transact.workflow.Timeout;
import dev.dbos.transact.workflow.VersionInfo;
import dev.dbos.transact.workflow.WorkflowDelay;
import dev.dbos.transact.workflow.WorkflowHandle;
import dev.dbos.transact.workflow.WorkflowSchedule;
import dev.dbos.transact.workflow.WorkflowStatus;

import java.time.Duration;
import java.time.Instant;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.UUID;

import javax.sql.DataSource;

import org.jspecify.annotations.NonNull;
import org.jspecify.annotations.Nullable;

/**
 * DBOSClient allows external programs to interact with DBOS apps via direct system database access.
 * Example interactions: Start/enqueue a workflow, and get the result Get events and send messages
 * to the workflow Manage workflows - list, fork, cancel, etc.
 */
public class DBOSClient implements AutoCloseable {
  private class WorkflowHandleClient<T, E extends Exception> implements WorkflowHandle<T, E> {
    private @NonNull String workflowId;

    public WorkflowHandleClient(@NonNull String workflowId) {
      this.workflowId = workflowId;
    }

    @Override
    public @NonNull String workflowId() {
      return workflowId;
    }

    @Override
    public T getResult() throws E {
      var result = systemDatabase.<T>awaitWorkflowResult(workflowId);
      return Result.<T, E>process(result);
    }

    @Override
    public @Nullable WorkflowStatus getStatus() {
      return systemDatabase.getWorkflowStatus(workflowId);
    }
  }

  private final @NonNull SystemDatabase systemDatabase;
  private final @Nullable DBOSSerializer serializer;

  /**
   * Construct a DBOSClient, by providing system database access credentials
   *
   * @param url System database JDBC URL
   * @param user System database user
   * @param password System database credential / password
   */
  public DBOSClient(@NonNull String url, @NonNull String user, @NonNull String password) {
    this(url, user, password, null, null, true);
  }

  /**
   * Construct a DBOSClient, by providing system database access credentials
   *
   * @param url System database JDBC URL
   * @param user System database user
   * @param password System database credential / password
   * @param schema Database schema for DBOS tables
   */
  public DBOSClient(
      @NonNull String url,
      @NonNull String user,
      @NonNull String password,
      @Nullable String schema) {
    this(url, user, password, schema, null, true);
  }

  /**
   * Construct a DBOSClient, by providing system database access credentials
   *
   * @param url System database JDBC URL
   * @param user System database user
   * @param password System database credential / password
   * @param schema Database schema for DBOS tables
   * @param serializer Custom serializer for serialization/deserialization
   */
  public DBOSClient(
      @NonNull String url,
      @NonNull String user,
      @NonNull String password,
      @Nullable String schema,
      @Nullable DBOSSerializer serializer) {
    this(url, user, password, schema, serializer, true);
  }

  /**
   * Construct a DBOSClient, by providing system database access credentials
   *
   * @param url System database JDBC URL
   * @param user System database user
   * @param password System database credential / password
   * @param schema Database schema for DBOS tables
   * @param serializer Custom serializer for serialization/deserialization
   * @param useListenNotify if true, use PostgreSQL LISTEN/NOTIFY for real-time event notifications
   */
  public DBOSClient(
      @NonNull String url,
      @NonNull String user,
      @NonNull String password,
      @Nullable String schema,
      @Nullable DBOSSerializer serializer,
      boolean useListenNotify) {
    this.serializer = serializer;
    systemDatabase = new SystemDatabase(url, user, password, schema, serializer, useListenNotify);
  }

  /**
   * Construct a DBOSClient, by providing a configured data source
   *
   * @param dataSource System database data source
   */
  public DBOSClient(@NonNull DataSource dataSource) {
    this(dataSource, null, null);
  }

  /**
   * Construct a DBOSClient, by providing a configured data source
   *
   * @param dataSource System database data source
   * @param schema Database schema for DBOS tables
   */
  public DBOSClient(@NonNull DataSource dataSource, @Nullable String schema) {
    this(dataSource, schema, null);
  }

  /**
   * Construct a DBOSClient, by providing a configured data source
   *
   * @param dataSource System database data source
   * @param schema Database schema for DBOS tables
   * @param serializer Custom serializer for serialization/deserialization
   */
  public DBOSClient(
      @NonNull DataSource dataSource,
      @Nullable String schema,
      @Nullable DBOSSerializer serializer) {
    this.serializer = serializer;
    systemDatabase = new SystemDatabase(dataSource, schema, serializer);
  }

  /**
   * Close this DBOSClient and release any underlying database resources. This method closes the
   * system database connection.
   */
  @Override
  public void close() {
    systemDatabase.close();
  }

  /**
   * Options for enqueuing a workflow instance for execution.
   *
   * <p>This record encapsulates all configuration required to enqueue a workflow, including:
   *
   * <ul>
   *   <li>Workflow name and (optionally) class and instance name
   *   <li>Target queue and queue-related options (priority, partitioning, deduplication, delay)
   *   <li>Workflow idempotency and versioning
   *   <li>Timeout and deadline management
   *   <li>Serialization strategy for workflow arguments
   * </ul>
   *
   * <p>Required fields: {@code workflowName}, {@code queueName}. All other fields are optional and
   * can be set using the provided {@code with} methods.
   *
   * @param workflowName The name of the workflow function to enqueue. Required.
   * @param className The Java class containing the workflow function. Optional.
   * @param instanceName The instance name for object-based workflows. Optional.
   * @param queueName The name of the queue to enqueue the workflow to. Required.
   * @param workflowId The idempotency key for the workflow instance. Optional; if not set, a random
   *     UUID will be generated.
   * @param appVersion The application version to target for execution. Optional.
   * @param timeout The maximum duration the workflow may run before being canceled. Optional.
   * @param deadline The absolute time by which the workflow must start or complete. Optional.
   * @param deduplicationId An optional ID to prevent duplicate enqueued workflows. Optional.
   * @param priority The priority to assign if the queue supports prioritization. Optional.
   * @param queuePartitionKey The partition key for distributing workflows across queue partitions.
   *     Optional.
   * @param delay The delay before the workflow starts executing. Optional.
   * @param serialization The serialization strategy for workflow arguments. Optional.
   */
  public record EnqueueOptions(
      @NonNull String workflowName,
      @Nullable String className,
      @Nullable String instanceName,
      @NonNull String queueName,
      @Nullable String workflowId,
      @Nullable String appVersion,
      @Nullable Duration timeout,
      @Nullable Instant deadline,
      @Nullable String deduplicationId,
      @Nullable Integer priority,
      @Nullable String queuePartitionKey,
      @Nullable Duration delay,
      @Nullable SerializationStrategy serialization) {

    public EnqueueOptions {
      if (nullableIsEmpty(workflowName)) {
        throw new IllegalArgumentException("workflowName must not be empty");
      }

      if (nullableIsEmpty(className)) {
        throw new IllegalArgumentException("className must not be empty");
      }

      if (nullableIsEmpty(instanceName)) {
        throw new IllegalArgumentException("instanceName must not be empty");
      }

      if (nullableIsEmpty(queueName)) {
        throw new IllegalArgumentException("queueName must not be empty");
      }

      if (nullableIsEmpty(workflowId)) {
        throw new IllegalArgumentException("workflowId must not be empty");
      }

      if (nullableIsEmpty(appVersion)) {
        throw new IllegalArgumentException("appVersion must not be empty");
      }

      if (nullableIsNotPositive(timeout)) {
        throw new IllegalArgumentException("timeout must be positive, non-zero duration");
      }

      if (nullableIsEmpty(deduplicationId)) {
        throw new IllegalArgumentException("deduplicationId must not be empty");
      }

      if (nullableIsEmpty(queuePartitionKey)) {
        throw new IllegalArgumentException("queuePartitionKey must not be empty");
      }

      if (nullableIsNotPositive(delay)) {
        throw new IllegalArgumentException("delay must be positive, non-zero duration");
      }
    }

    /** Construct `EnqueueOptions` with a minimum set of required options */
    public EnqueueOptions(@NonNull String workflowName, @NonNull String queueName) {
      this(
          workflowName,
          null,
          null,
          queueName,
          null,
          null,
          null,
          null,
          null,
          null,
          null,
          null,
          null);
    }

    public EnqueueOptions(
        @NonNull String workflowName, @Nullable String className, @NonNull String queueName) {
      this(
          workflowName,
          className,
          null,
          queueName,
          null,
          null,
          null,
          null,
          null,
          null,
          null,
          null,
          null);
    }

    /**
     * Specify the Java classname for the class containing the workflow to enqueue
     *
     * @param className Class containing the workflow to enqueue
     * @return New `EnqueueOptions` with the class name set
     */
    public @NonNull EnqueueOptions withClassName(@Nullable String className) {
      return new EnqueueOptions(
          this.workflowName,
          className,
          this.instanceName,
          this.queueName,
          this.workflowId,
          this.appVersion,
          this.timeout,
          this.deadline,
          this.deduplicationId,
          this.priority,
          this.queuePartitionKey,
          this.delay,
          this.serialization);
    }

    /**
     * Specify the workflow ID for the workflow to be enqueued. This is an idempotency key for
     * running the workflow.
     *
     * @param workflowId Workflow idempotency ID to use
     * @return New `EnqueueOptions` with the workflow ID set
     */
    public @NonNull EnqueueOptions withWorkflowId(@Nullable String workflowId) {
      return new EnqueueOptions(
          this.workflowName,
          this.className,
          this.instanceName,
          this.queueName,
          workflowId,
          this.appVersion,
          this.timeout,
          this.deadline,
          this.deduplicationId,
          this.priority,
          this.queuePartitionKey,
          this.delay,
          this.serialization);
    }

    /**
     * Specify the app version for the workflow to be enqueued. The workflow will be executed by an
     * executor with this app version. If not specified, the current app version will be used.
     *
     * @param appVersion Application version to use for executing the workflow
     * @return New `EnqueueOptions` with the app version set
     */
    public @NonNull EnqueueOptions withAppVersion(@Nullable String appVersion) {
      return new EnqueueOptions(
          this.workflowName,
          this.className,
          this.instanceName,
          this.queueName,
          this.workflowId,
          appVersion,
          this.timeout,
          this.deadline,
          this.deduplicationId,
          this.priority,
          this.queuePartitionKey,
          this.delay,
          this.serialization);
    }

    /**
     * Specify a timeout for the workflow to be enqueued. Timeout begins once the workflow is
     * running; if it exceeds this it will be canceled.
     *
     * @param timeout Duration of time, from start, before the workflow is canceled.
     * @return New `EnqueueOptions` with the timeout set
     */
    public @NonNull EnqueueOptions withTimeout(@Nullable Duration timeout) {
      return new EnqueueOptions(
          this.workflowName,
          this.className,
          this.instanceName,
          this.queueName,
          this.workflowId,
          this.appVersion,
          timeout,
          this.deadline,
          this.deduplicationId,
          this.priority,
          this.queuePartitionKey,
          this.delay,
          this.serialization);
    }

    /**
     * Specify a deadline for the workflow. This is an absolute time, regardless of when the
     * workflow starts.
     *
     * @param deadline Instant after which the workflow will be canceled.
     * @return New `EnqueueOptions` with the deadline set
     */
    public @NonNull EnqueueOptions withDeadline(@Nullable Instant deadline) {
      return new EnqueueOptions(
          this.workflowName,
          this.className,
          this.instanceName,
          this.queueName,
          this.workflowId,
          this.appVersion,
          this.timeout,
          deadline,
          this.deduplicationId,
          this.priority,
          this.queuePartitionKey,
          this.delay,
          this.serialization);
    }

    /**
     * Specify a queue deduplication ID for the workflow to be enqueued. Queue requests with the
     * same deduplication ID will be rejected.
     *
     * @param deduplicationId Queue deduplication ID
     * @return New `EnqueueOptions` with the deduplication ID set
     */
    public @NonNull EnqueueOptions withDeduplicationId(@Nullable String deduplicationId) {
      return new EnqueueOptions(
          this.workflowName,
          this.className,
          this.instanceName,
          this.queueName,
          this.workflowId,
          this.appVersion,
          this.timeout,
          this.deadline,
          deduplicationId,
          this.priority,
          this.queuePartitionKey,
          this.delay,
          this.serialization);
    }

    /**
     * Specify an object instance name to execute the workflow. If workflow objects are named, this
     * must be specified to direct processing to the correct instance.
     *
     * @param instName Instance name registered within `DBOS.registerWorkflows`
     * @return New `EnqueueOptions` with the target instance name set
     */
    public @NonNull EnqueueOptions withInstanceName(@Nullable String instName) {
      return new EnqueueOptions(
          this.workflowName,
          this.className,
          instName,
          this.queueName,
          this.workflowId,
          this.appVersion,
          this.timeout,
          this.deadline,
          this.deduplicationId,
          this.priority,
          this.queuePartitionKey,
          this.delay,
          this.serialization);
    }

    /**
     * Specify priority. Priority must be enabled on the queue for this to be effective.
     *
     * @param priority Queue priority; if `null`, priority '0' will be used.
     * @return New `EnqueueOptions` with the priority set
     */
    public @NonNull EnqueueOptions withPriority(@Nullable Integer priority) {
      return new EnqueueOptions(
          this.workflowName,
          this.className,
          this.instanceName,
          this.queueName,
          this.workflowId,
          this.appVersion,
          this.timeout,
          this.deadline,
          this.deduplicationId,
          priority,
          this.queuePartitionKey,
          this.delay,
          this.serialization);
    }

    /**
     * Creates a new EnqueueOptions instance with the specified queue partition key. The partition
     * key is used to determine which partition of the queue the workflow should be enqueued to,
     * allowing for better load distribution and ordering guarantees.
     *
     * @param partitionKey the partition key to use for queue partitioning, can be null
     * @return a new EnqueueOptions instance with the specified partition key
     */
    public @NonNull EnqueueOptions withQueuePartitionKey(@Nullable String partitionKey) {
      return new EnqueueOptions(
          this.workflowName,
          this.className,
          this.instanceName,
          this.queueName,
          this.workflowId,
          this.appVersion,
          this.timeout,
          this.deadline,
          this.deduplicationId,
          this.priority,
          partitionKey,
          this.delay,
          this.serialization);
    }

    /**
     * Specify a delay before the workflow starts executing. The workflow will remain in the queue
     * until the delay has elapsed.
     *
     * @param delay Duration to wait before the workflow begins execution.
     * @return New `EnqueueOptions` with the delay set
     */
    public @NonNull EnqueueOptions withDelay(@Nullable Duration delay) {
      return new EnqueueOptions(
          this.workflowName,
          this.className,
          this.instanceName,
          this.queueName,
          this.workflowId,
          this.appVersion,
          this.timeout,
          this.deadline,
          this.deduplicationId,
          this.priority,
          this.queuePartitionKey,
          delay,
          this.serialization);
    }

    /**
     * Specify the serialization strategy for the workflow arguments.
     *
     * @param serialization The serialization strategy ({@link SerializationStrategy#PORTABLE} for
     *     cross-language compatibility, {@link SerializationStrategy#NATIVE} for Java-specific, or
     *     {@link SerializationStrategy#DEFAULT} for the default behavior)
     * @return New `EnqueueOptions` with the serialization strategy set
     */
    public @NonNull EnqueueOptions withSerialization(
        @Nullable SerializationStrategy serialization) {
      return new EnqueueOptions(
          this.workflowName,
          this.className,
          this.instanceName,
          this.queueName,
          this.workflowId,
          this.appVersion,
          this.timeout,
          this.deadline,
          this.deduplicationId,
          this.priority,
          this.queuePartitionKey,
          this.delay,
          serialization);
    }
  }

  /**
   * Enqueue a workflow with explicit serialization format and support for both positional and named
   * arguments.
   *
   * @param <T> Return type of workflow function
   * @param <E> Exception thrown by workflow function
   * @param options {@link EnqueueOptions} for configuring the workflow enqueue
   * @param positionalArgs Positional arguments to pass to the workflow function
   * @param namedArgs Named arguments to pass to the workflow function (e.g., for Python kwargs)
   * @param serializationFormat Serialization format string to use (null for default)
   * @return WorkflowHandle for retrieving workflow ID, status, and results
   */
  public <T, E extends Exception> @NonNull WorkflowHandle<T, E> enqueueWorkflow(
      @NonNull EnqueueOptions options,
      @Nullable Object[] positionalArgs,
      @Nullable Map<String, Object> namedArgs,
      @Nullable String serializationFormat) {

    Objects.requireNonNull(options, "options must not be null");
    Objects.requireNonNull(options.workflowName, "EnqueueOptions workflowName must not be null");
    Objects.requireNonNull(options.queueName, "EnqueueOptions queueName must not be null");

    if (options.timeout != null && options.deadline != null) {
      throw new IllegalArgumentException("Can't set timeout and deadline EnqueueOptions");
    }

    var workflowId =
        Objects.requireNonNullElseGet(options.workflowId(), () -> UUID.randomUUID().toString());

    DBOSExecutor.enqueueWorkflow(
        options.workflowName(),
        options.className(),
        options.instanceName(),
        null,
        positionalArgs,
        namedArgs,
        new ExecutionOptions(
            workflowId,
            Timeout.of(options.timeout()),
            options.deadline,
            options.queueName(),
            options.deduplicationId,
            options.priority,
            options.queuePartitionKey,
            options.delay,
            options.appVersion,
            false,
            false,
            serializationFormat),
        null,
        null,
        null,
        null,
        systemDatabase,
        this.serializer);

    return new WorkflowHandleClient<>(workflowId);
  }

  /**
   * Enqueue a workflow.
   *
   * @param <T> Return type of workflow function
   * @param <E> Exception thrown by workflow function
   * @param options `DBOSClient.EnqueueOptions` for enqueuing the workflow
   * @param args Arguments to pass to the workflow function
   * @return WorkflowHandle for retrieving workflow ID, status, and results
   */
  public <T, E extends Exception> @NonNull WorkflowHandle<T, E> enqueueWorkflow(
      @NonNull EnqueueOptions options, @Nullable Object[] args) {
    var serializationFormat =
        options.serialization() != null ? options.serialization().formatName() : null;
    return enqueueWorkflow(options, args, null, serializationFormat);
  }

  /**
   * Enqueue a workflow using portable JSON serialization. This method is intended for
   * cross-language workflow initiation where the workflow function definition may not be available
   * in Java.
   *
   * @param <T> Return type of workflow function
   * @param options `DBOSClient.EnqueueOptions` for enqueuing the workflow
   * @param positionalArgs Positional arguments to pass to the workflow function
   * @param namedArgs Optional named arguments (for workflows that support them, e.g., Python
   *     kwargs)
   * @return WorkflowHandle for retrieving workflow ID, status, and results
   */
  public <T> @NonNull WorkflowHandle<T, PortableWorkflowException> enqueuePortableWorkflow(
      @NonNull EnqueueOptions options,
      @Nullable Object[] positionalArgs,
      @Nullable Map<String, Object> namedArgs) {
    return enqueueWorkflow(options, positionalArgs, namedArgs, SerializationUtil.PORTABLE);
  }

  /** Options for sending a message. */
  public record SendOptions(@Nullable SerializationStrategy serialization, boolean sendToForks) {
    /**
     * Create SendOptions with default serialization strategy. Uses the system's default
     * serialization format for message encoding.
     *
     * @return SendOptions configured with default serialization
     */
    public static SendOptions defaults() {
      return new SendOptions(SerializationStrategy.DEFAULT, false);
    }

    /**
     * Create SendOptions with portable JSON serialization strategy. Uses portable JSON format
     * suitable for cross-language workflow communication.
     *
     * @return SendOptions configured with portable JSON serialization
     */
    public static SendOptions portable() {
      return new SendOptions(SerializationStrategy.PORTABLE, false);
    }

    /**
     * Create a new SendOptions with the sendToForks flag set.
     *
     * @param v if true, deliver the message to any forked copies of the destination workflow
     * @return new SendOptions with the sendToForks flag updated
     */
    public SendOptions withSendToForks(boolean v) {
      return new SendOptions(serialization, v);
    }
  }

  /**
   * Send a message to a workflow
   *
   * @param destinationId workflowId of the workflow to receive the message
   * @param message Message contents
   * @param topic Topic for the message
   * @param idempotencyKey If specified, use the value to ensure exactly-once send semantics
   */
  public void send(
      @NonNull String destinationId,
      @NonNull Object message,
      @NonNull String topic,
      @Nullable String idempotencyKey) {
    send(destinationId, message, topic, idempotencyKey, null);
  }

  /**
   * Send a message to a workflow with serialization options
   *
   * @param destinationId workflowId of the workflow to receive the message
   * @param message Message contents
   * @param topic Topic for the message
   * @param idempotencyKey If specified, use the value to ensure exactly-once send semantics
   * @param options Optional send options including serialization type
   */
  public void send(
      @NonNull String destinationId,
      @NonNull Object message,
      @NonNull String topic,
      @Nullable String idempotencyKey,
      @Nullable SendOptions options) {

    String serializationFormat =
        (options != null && options.serialization() != null)
            ? options.serialization().formatName()
            : null;
    boolean sendToForks = options != null && options.sendToForks();

    systemDatabase.sendBulk(
        List.of(new SendMessage(destinationId, message, topic, idempotencyKey)),
        null,
        -1,
        "DBOS.send",
        sendToForks,
        serializationFormat);
  }

  /**
   * Send multiple messages to workflows using default options
   *
   * @param messages list of messages to send
   */
  public void sendBulk(@NonNull List<SendMessage> messages) {
    sendBulk(messages, null);
  }

  /**
   * Send multiple messages to workflows with serialization options
   *
   * @param messages list of messages to send
   * @param options optional send options including serialization type and fork delivery; null for
   *     defaults
   */
  public void sendBulk(@NonNull List<SendMessage> messages, @Nullable SendOptions options) {
    String serializationFormat =
        (options != null && options.serialization() != null)
            ? options.serialization().formatName()
            : null;
    boolean sendToForks = options != null && options.sendToForks();
    systemDatabase.sendBulk(messages, null, -1, "DBOS.sendBulk", sendToForks, serializationFormat);
  }

  /**
   * Get event from a workflow
   *
   * @param targetId ID of the workflow setting the event
   * @param key Key for the event
   * @param timeout Maximum time duration to wait before timing out
   * @return Optional containing the workflow event value if available, or empty if timeout occurs
   *     or no event found
   */
  public @NonNull Optional<Object> getEvent(
      @NonNull String targetId, @NonNull String key, @NonNull Duration timeout) {
    return Optional.ofNullable(systemDatabase.getEvent(targetId, key, timeout, null));
  }

  /**
   * Read values from a stream as an iterator. This function reads values from a stream identified
   * by the workflow_id and key, returning an iterator that yields each value in order until the
   * stream is closed or the workflow terminates.
   *
   * @param workflowId The workflow instance ID that owns the stream
   * @param key The stream key / name within the workflow
   * @return Iterator that yields each value in the stream
   */
  public @NonNull Iterator<Object> readStream(@NonNull String workflowId, @NonNull String key) {
    return new StreamIterator(workflowId, key, systemDatabase);
  }

  /**
   * Create a handle for a workflow. This call does not ensure that the workflow exists; use the
   * returned handle's `getStatus()`.
   *
   * @param <T> Type of the workflow's return value
   * @param <E> Type of any checked exception thrown by the workflow
   * @param workflowId ID of the workflow to retrieve
   * @return A `WorkflowHandle` for the specified worflow ID
   */
  public <T, E extends Exception> @NonNull WorkflowHandle<T, E> retrieveWorkflow(
      @NonNull String workflowId) {
    return new WorkflowHandleClient<>(workflowId);
  }

  /**
   * Cancel a worflow
   *
   * @param workflowId ID of the workflow to cancel
   */
  public void cancelWorkflow(@NonNull String workflowId) {
    systemDatabase.cancelWorkflows(List.of(workflowId));
  }

  /**
   * Cancel multiple workflows. After this function is called, the next step (not the current one)
   * of each specified workflow will not execute.
   *
   * @param workflowIds a list of workflow IDs to cancel; must not be null
   */
  public void cancelWorkflows(@NonNull List<String> workflowIds) {
    systemDatabase.cancelWorkflows(workflowIds);
  }

  /**
   * Resume a workflow starting from the step after the last complete step. This method allows
   * resuming workflows that were previously interrupted, failed, or canceled. The workflow will
   * continue execution from where it left off, replaying any completed steps deterministically.
   *
   * @param <T> Type of the workflow's return value
   * @param <E> Type of any checked exception thrown by the workflow
   * @param workflowId ID of the workflow to resume; must not be null
   * @param queueName optional queue name to enqueue the resumed workflow to; if null, the workflow
   *     will be resumed in the default execution context
   * @return WorkflowHandle for the resumed workflow
   */
  public <T, E extends Exception> @NonNull WorkflowHandle<T, E> resumeWorkflow(
      @NonNull String workflowId, @Nullable String queueName) {
    systemDatabase.resumeWorkflows(List.of(workflowId), queueName);
    return retrieveWorkflow(workflowId);
  }

  /**
   * Resume a workflow starting from the step after the last complete step using the default queue.
   * This method is equivalent to calling {@code resumeWorkflow(workflowId, null)}. The workflow
   * will continue execution from where it left off, replaying any completed steps
   * deterministically.
   *
   * @param <T> Type of the workflow's return value
   * @param <E> Type of any checked exception thrown by the workflow
   * @param workflowId ID of the workflow to resume; must not be null
   * @return WorkflowHandle for the resumed workflow
   */
  public <T, E extends Exception> @NonNull WorkflowHandle<T, E> resumeWorkflow(
      @NonNull String workflowId) {
    return resumeWorkflow(workflowId, null);
  }

  /**
   * Resume multiple workflows starting from the step after the last complete step for each workflow
   * using the default queue. This method is equivalent to calling {@code
   * resumeWorkflows(workflowIds, null)}. Each workflow will continue execution from where it left
   * off, replaying any completed steps deterministically.
   *
   * @param workflowIds a list of workflow IDs to resume; must not be null
   * @return A list of handles to the resumed workflows
   */
  public @NonNull List<WorkflowHandle<Object, Exception>> resumeWorkflows(
      @NonNull List<String> workflowIds) {
    return resumeWorkflows(workflowIds, null);
  }

  /**
   * Resume multiple workflows starting from the step after the last complete step for each
   * workflow. This method allows bulk resumption of workflows that were previously interrupted,
   * failed, or canceled. Each workflow will continue execution from where it left off, replaying
   * any completed steps deterministically.
   *
   * @param workflowIds a list of workflow IDs to resume; must not be null
   * @param queueName optional queue name to enqueue the resumed workflows to; if null, the
   *     workflows will be resumed in the default execution context
   * @return A list of handles to the resumed workflows
   */
  public @NonNull List<WorkflowHandle<Object, Exception>> resumeWorkflows(
      @NonNull List<String> workflowIds, @Nullable String queueName) {
    systemDatabase.resumeWorkflows(workflowIds, queueName);
    return workflowIds.stream().map(this::retrieveWorkflow).toList();
  }

  /**
   * Delete a workflow from the system. This permanently removes the workflow and its associated
   * data from the database. Child workflows are preserved by default.
   *
   * @param workflowId ID of the workflow to delete; must not be null
   */
  public void deleteWorkflow(@NonNull String workflowId) {
    deleteWorkflows(List.of(workflowId), false);
  }

  /**
   * Delete a workflow from the system. This permanently removes the workflow and its associated
   * data from the database.
   *
   * @param workflowId ID of the workflow to delete; must not be null
   * @param deleteChildren if true, also delete any child workflows; if false, preserve child
   *     workflows
   */
  public void deleteWorkflow(@NonNull String workflowId, boolean deleteChildren) {
    deleteWorkflows(List.of(workflowId), deleteChildren);
  }

  /**
   * Delete multiple workflows from the system. This permanently removes the workflows and their
   * associated data from the database. Child workflows are preserved by default.
   *
   * @param workflowIds a list of workflow IDs to delete; must not be null
   */
  public void deleteWorkflows(@NonNull List<String> workflowIds) {
    deleteWorkflows(workflowIds, false);
  }

  /**
   * Delete multiple workflows from the system. This permanently removes the workflows and their
   * associated data from the database.
   *
   * @param workflowIds a list of workflow IDs to delete; must not be null
   * @param deleteChildren if true, also delete any child workflows; if false, preserve child
   *     workflows
   */
  public void deleteWorkflows(@NonNull List<String> workflowIds, boolean deleteChildren) {
    systemDatabase.deleteWorkflows(workflowIds, deleteChildren);
  }

  /**
   * Fork a workflow, providing a handle to the new workflow
   *
   * @param <T> Type of the workflow's return value
   * @param <E> Type of any checked exception thrown by the workflow
   * @param originalWorkflowId ID of the workflow to fork
   * @param startStep Step number for starting the new fork of the workflow; if zero start from the
   *     beginning
   * @param options Options for forking;
   * @return `WorkflowHandle` for the new workflow
   */
  public <T, E extends Exception> @NonNull WorkflowHandle<T, E> forkWorkflow(
      @NonNull String originalWorkflowId, int startStep, @NonNull ForkOptions options) {
    var forkedWorkflowId = systemDatabase.forkWorkflow(originalWorkflowId, startStep, options);
    return retrieveWorkflow(forkedWorkflowId);
  }

  /**
   * Get the status of a workflow
   *
   * @param workflowId ID of the workflow to query for status
   * @return WorkflowStatus of the workflow, or empty if the workflow does not exist
   */
  public @NonNull Optional<WorkflowStatus> getWorkflowStatus(@NonNull String workflowId) {
    return Optional.ofNullable(systemDatabase.getWorkflowStatus(workflowId));
  }

  /**
   * List workflows matching the supplied input filter criteria
   *
   * @param input Filter criteria to use for listing workflows. Pass null to list all workflows.
   * @return list of workflows matching the `ListWorkflowsInput` criteria
   */
  public @NonNull List<WorkflowStatus> listWorkflows(@Nullable ListWorkflowsInput input) {
    return systemDatabase.listWorkflows(input);
  }

  /**
   * List the steps executed by a workflow
   *
   * @param workflowId ID of the workflow to list
   * @return List of steps executed by the workflow
   */
  public @NonNull List<StepInfo> listWorkflowSteps(@NonNull String workflowId) {
    return listWorkflowSteps(workflowId, null, null);
  }

  /**
   * List the steps executed by a workflow with optional pagination
   *
   * @param workflowId ID of the workflow to list
   * @param limit Maximum number of steps to return
   * @param offset Number of steps to skip before returning
   * @return List of steps executed by the workflow
   */
  public @NonNull List<StepInfo> listWorkflowSteps(
      @NonNull String workflowId, Integer limit, Integer offset) {
    return systemDatabase.listWorkflowSteps(workflowId, true, limit, offset);
  }

  /**
   * List all registered application versions, ordered by timestamp descending.
   *
   * @return list of {@link VersionInfo} records
   */
  public @NonNull List<VersionInfo> listApplicationVersions() {
    return systemDatabase.listApplicationVersions();
  }

  /**
   * Get the most recently promoted application version.
   *
   * @return the latest {@link VersionInfo}
   */
  public @NonNull VersionInfo getLatestApplicationVersion() {
    return systemDatabase.getLatestApplicationVersion();
  }

  /**
   * Promote an existing version to be the latest application version by updating its timestamp.
   *
   * @param versionName the version to promote; it must already exist
   */
  public void setLatestApplicationVersion(@NonNull String versionName) {
    systemDatabase.updateApplicationVersionTimestamp(versionName, Instant.now());
  }

  /**
   * Create a cron schedule. The scheduleId is generated if null.
   *
   * @param schedule the schedule configuration
   */
  public void createSchedule(@NonNull WorkflowSchedule schedule) {
    systemDatabase.createSchedule(schedule);
  }

  /**
   * Get a schedule by name.
   *
   * @param name schedule name
   * @return the schedule, or empty if not found
   */
  public @NonNull Optional<WorkflowSchedule> getSchedule(@NonNull String name) {
    return systemDatabase.getSchedule(name);
  }

  /**
   * List schedules with optional filters.
   *
   * @param status filter by status; null means no filter
   * @param workflowName filter by workflow name; null means no filter
   * @param namePrefix filter by schedule name prefix; null means no filter
   * @return matching schedules
   */
  public @NonNull List<WorkflowSchedule> listSchedules(
      @Nullable List<ScheduleStatus> status,
      @Nullable List<String> workflowName,
      @Nullable List<String> namePrefix) {
    return systemDatabase.listSchedules(status, workflowName, namePrefix);
  }

  /**
   * Delete a schedule by name. No-op if the schedule does not exist.
   *
   * @param name schedule name
   */
  public void deleteSchedule(@NonNull String name) {
    systemDatabase.deleteSchedule(name);
  }

  /**
   * Pause a schedule. A paused schedule does not fire.
   *
   * @param name schedule name
   */
  public void pauseSchedule(@NonNull String name) {
    systemDatabase.pauseSchedule(name);
  }

  /**
   * Resume a paused schedule so it begins firing again.
   *
   * @param name schedule name
   */
  public void resumeSchedule(@NonNull String name) {
    systemDatabase.resumeSchedule(name);
  }

  /**
   * Atomically create or replace a set of schedules.
   *
   * @param schedules the schedules to apply
   */
  public void applySchedules(@NonNull List<@NonNull WorkflowSchedule> schedules) {
    systemDatabase.applySchedules(schedules);
  }

  /**
   * Atomically create or replace a set of schedules.
   *
   * @param schedules the schedules to apply
   */
  public void applySchedules(@NonNull WorkflowSchedule... schedules) {
    systemDatabase.applySchedules(Arrays.asList(schedules));
  }

  /**
   * Enqueue all executions of a schedule that would have run between {@code start} (exclusive) and
   * {@code end} (exclusive).
   *
   * @param scheduleName name of an existing schedule
   * @param start start of the backfill window (exclusive)
   * @param end end of the backfill window (exclusive)
   * @return handles to the enqueued executions
   */
  public @NonNull List<WorkflowHandle<Object, Exception>> backfillSchedule(
      @NonNull String scheduleName, @NonNull Instant start, @NonNull Instant end) {
    var ids = DBOSExecutor.backfillSchedule(scheduleName, start, end, systemDatabase, serializer);
    return ids.stream().<WorkflowHandle<Object, Exception>>map(this::retrieveWorkflow).toList();
  }

  /**
   * Immediately enqueue the scheduled workflow at the current time.
   *
   * @param scheduleName name of an existing schedule
   * @return handle to the enqueued execution
   */
  public <T, E extends Exception> @NonNull WorkflowHandle<T, E> triggerSchedule(
      @NonNull String scheduleName) {
    var id = DBOSExecutor.triggerSchedule(scheduleName, systemDatabase, serializer);
    return retrieveWorkflow(id);
  }

  /**
   * Sets a delay for a workflow, causing it to be paused for a specified duration or until a
   * specific time. This is useful for implementing delays, timeouts, or scheduling workflows to
   * resume at a later time.
   *
   * @param workflowId the unique identifier of the workflow to delay
   * @param delay the duration to delay the workflow from now
   * @throws IllegalArgumentException if the workflow ID is invalid
   */
  public void setWorkflowDelay(@NonNull String workflowId, @NonNull Duration delay) {
    var wfDelay = new WorkflowDelay.Delay(Objects.requireNonNull(delay, "delay must not be null"));
    systemDatabase.setWorkflowDelay(workflowId, wfDelay);
  }

  /**
   * Sets a delay for a workflow, causing it to be paused for a specified duration or until a
   * specific time. This is useful for implementing delays, timeouts, or scheduling workflows to
   * resume at a later time.
   *
   * @param workflowId the unique identifier of the workflow to delay
   * @param delayUntil the absolute time until which to delay the workflow
   * @throws IllegalArgumentException if the workflow ID is invalid
   */
  public void setWorkflowDelay(@NonNull String workflowId, @NonNull Instant delayUntil) {
    var wfDelay =
        new WorkflowDelay.DelayUntil(
            Objects.requireNonNull(delayUntil, "delayUntil must not be null"));
    systemDatabase.setWorkflowDelay(workflowId, wfDelay);
  }

  /**
   * Register a database-backed dynamic queue. Uses {@link QueueConflictResolution#ALWAYS_UPDATE} by
   * default.
   *
   * @param name Queue name
   * @param options Configuration options
   */
  public void registerQueue(@NonNull String name, @NonNull QueueOptions options) {
    registerQueue(name, options, QueueConflictResolution.ALWAYS_UPDATE);
  }

  /**
   * Register a database-backed dynamic queue.
   *
   * <p>{@link QueueConflictResolution#UPDATE_IF_LATEST_VERSION} is not supported by {@code
   * DBOSClient} because clients are not associated with an application version. Use {@link
   * QueueConflictResolution#ALWAYS_UPDATE} or {@link QueueConflictResolution#NEVER_UPDATE}.
   *
   * @param name Queue name
   * @param options Configuration options
   * @param onConflict How to handle an existing queue with the same name
   */
  public void registerQueue(
      @NonNull String name,
      @NonNull QueueOptions options,
      @NonNull QueueConflictResolution onConflict) {
    if (onConflict == QueueConflictResolution.UPDATE_IF_LATEST_VERSION) {
      throw new IllegalArgumentException(
          "DBOSClient.registerQueue does not support UPDATE_IF_LATEST_VERSION because clients are"
              + " not associated with an application version. Use ALWAYS_UPDATE or NEVER_UPDATE.");
    }
    systemDatabase.upsertQueue(name, options, onConflict == QueueConflictResolution.ALWAYS_UPDATE);
  }

  /**
   * Update the configuration of a database-backed dynamic queue. Only fields that are present in
   * {@code options} are written; absent fields are left unchanged.
   *
   * @param name Queue name
   * @param options Fields to update
   */
  public void updateQueue(@NonNull String name, @NonNull QueueOptions options) {
    systemDatabase.updateQueue(name, options);
  }

  /**
   * Retrieve a database-backed dynamic queue by name.
   *
   * @param name Queue name
   * @return the queue if it exists in the database, or empty
   */
  public @NonNull Optional<Queue> findQueue(@NonNull String name) {
    return systemDatabase.findQueue(name);
  }

  /**
   * List all database-backed dynamic queues.
   *
   * @return list of all queues currently registered in the database
   */
  public @NonNull List<Queue> listQueues() {
    return systemDatabase.listQueues();
  }

  /**
   * Delete a database-backed dynamic queue.
   *
   * @param name Queue name
   * @return true if the queue was deleted, false if it did not exist
   */
  public boolean deleteQueue(@NonNull String name) {
    return systemDatabase.deleteQueue(name);
  }
}