Workflow Implementation and Lifecycle

/**
 * Central utility class providing workflow execution context and APIs for activities, timers, signals, etc.
 */
public final class Workflow {
    /**
     * Creates activity client stub with default options.
     * @param activityInterface activity interface class
     * @return activity stub
     */
    public static <T> T newActivityStub(Class<T> activityInterface);

    /**
     * Creates activity client stub with options.
     * @param activityInterface activity interface class
     * @param options activity options
     * @return activity stub
     */
    public static <T> T newActivityStub(Class<T> activityInterface, ActivityOptions options);

    /**
     * Creates local activity stub with options.
     * @param activityInterface activity interface class
     * @param options local activity options
     * @return local activity stub
     */
    public static <T> T newLocalActivityStub(Class<T> activityInterface, LocalActivityOptions options);

    /**
     * Creates child workflow stub.
     * @param workflowInterface workflow interface class
     * @param options child workflow options
     * @return child workflow stub
     */
    public static <T> T newChildWorkflowStub(Class<T> workflowInterface, ChildWorkflowOptions options);

    /**
     * Creates external workflow stub for signaling.
     * @param workflowInterface workflow interface class
     * @param workflowId target workflow ID
     * @return external workflow stub
     */
    public static <T> T newExternalWorkflowStub(Class<T> workflowInterface, String workflowId);

    /**
     * Creates external workflow stub with run ID.
     * @param workflowInterface workflow interface class
     * @param workflowExecution target workflow execution
     * @return external workflow stub
     */
    public static <T> T newExternalWorkflowStub(Class<T> workflowInterface, WorkflowExecution workflowExecution);

    /**
     * Deterministic sleep/timer.
     * @param duration sleep duration
     */
    public static void sleep(Duration duration);

    /**
     * Deterministic current time.
     * @return current time in milliseconds since epoch
     */
    public static long currentTimeMillis();

    /**
     * Deterministic random number generator.
     * @return random instance seeded with workflow execution
     */
    public static Random newRandom();

    /**
     * Execute side effect that is recorded in workflow history.
     * @param resultClass result class
     * @param func side effect function
     * @return side effect result
     */
    public static <R> R sideEffect(Class<R> resultClass, Func<R> func);

    /**
     * Execute mutable side effect with change detection.
     * @param id unique side effect identifier
     * @param resultClass result class
     * @param updated function to compare old and new values
     * @param func side effect function
     * @return side effect result
     */
    public static <R> R mutableSideEffect(String id, Class<R> resultClass, BiPredicate<R, R> updated, Func<R> func);

    /**
     * Get workflow version for backwards compatibility.
     * @param changeId unique change identifier
     * @param minSupported minimum supported version
     * @param maxSupported maximum supported version
     * @return workflow version
     */
    public static int getVersion(String changeId, int minSupported, int maxSupported);

    /**
     * Continue as new workflow with same type.
     * @param args new workflow arguments
     */
    public static void continueAsNew(Object... args);

    /**
     * Continue as new workflow with options.
     * @param options continue as new options
     * @param args new workflow arguments
     */
    public static void continueAsNew(ContinueAsNewOptions options, Object... args);

    /**
     * Create cancellation scope for managing cancellation.
     * @param detached whether scope is detached from parent
     * @param runnable code to execute in scope
     * @return cancellation scope
     */
    public static CancellationScope newCancellationScope(boolean detached, Runnable runnable);

    /**
     * Create disconnected cancellation scope.
     * @param runnable code to execute in scope
     * @return disconnected cancellation scope
     */
    public static CancellationScope newDetachedCancellationScope(Runnable runnable);

    /**
     * Get current cancellation scope.
     * @return current cancellation scope
     */
    public static CancellationScope current();

    /**
     * Returns workflow info for current execution.
     * @return workflow info
     */
    public static WorkflowInfo getInfo();

    /**
     * Returns logger for current workflow.
     * @param clazz logger class
     * @return logger instance
     */
    public static Logger getLogger(Class<?> clazz);

    /**
     * Returns logger for current workflow.
     * @param name logger name
     * @return logger instance
     */
    public static Logger getLogger(String name);

    /**
     * Returns metrics scope for workflow metrics.
     * @return metrics scope
     */
    public static Scope getMetricsScope();

    /**
     * Get search attributes for current workflow.
     * @return search attributes
     */
    public static SearchAttributes getSearchAttributes();

    /**
     * Get typed search attributes for current workflow.
     * @return typed search attributes
     */
    public static SearchAttributes getTypedSearchAttributes();

    /**
     * Upsert search attributes for current workflow.
     * @param searchAttributes search attributes to upsert
     */
    public static void upsertSearchAttributes(Map<String, ?> searchAttributes);

    /**
     * Upsert typed search attributes for current workflow.
     * @param searchAttributes typed search attributes to upsert
     */
    public static void upsertTypedSearchAttributes(SearchAttributes searchAttributes);

    /**
     * Get memo for current workflow.
     * @return memo map
     */
    public static Map<String, Object> getMemo();

    /**
     * Register query handler dynamically.
     * @param queryType query type name
     * @param queryHandler query handler function
     */
    public static <R> void registerListener(String queryType, Functions.Func<R> queryHandler);

    /**
     * Register signal handler dynamically.
     * @param signalType signal type name
     * @param signalHandler signal handler function
     */
    public static void registerListener(String signalType, Functions.Proc1<Object> signalHandler);

    /**
     * Register update handler dynamically.
     * @param updateType update type name
     * @param updateHandler update handler function
     */
    public static <R> void registerListener(String updateType, Functions.Func1<Object, R> updateHandler);

    /**
     * Await condition with timeout.
     * @param timeout maximum wait time
     * @param condition condition supplier
     * @return true if condition became true within timeout
     */
    public static boolean await(Duration timeout, Supplier<Boolean> condition);

    /**
     * Await condition indefinitely.
     * @param condition condition supplier
     */
    public static void await(Supplier<Boolean> condition);

    /**
     * Execute promise asynchronously.
     * @param task async task
     * @return promise for async execution
     */
    public static <R> Promise<R> async(Functions.Func<R> task);

    /**
     * Execute procedure asynchronously.
     * @param task async procedure
     * @return promise for async execution
     */
    public static Promise<Void> async(Functions.Proc task);

    /**
     * Create new promise.
     * @return new promise
     */
    public static <R> Promise<R> newPromise();

    /**
     * Wait for all promises to complete.
     * @param promises collection of promises
     * @return promise that completes when all input promises complete
     */
    public static Promise<Void> allOf(Promise<?>... promises);

    /**
     * Wait for any promise to complete.
     * @param promises collection of promises
     * @return promise that completes when any input promise completes
     */
    public static Promise<Object> anyOf(Promise<?>... promises);
}

@WorkflowInterface
public interface OrderProcessingWorkflow {
    @WorkflowMethod
    OrderResult processOrder(OrderRequest order);

    @SignalMethod
    void addItem(OrderItem item);

    @QueryMethod
    OrderStatus getStatus();

    @UpdateMethod
    void updateShippingAddress(Address newAddress);
}

public class OrderProcessingWorkflowImpl implements OrderProcessingWorkflow {
    private final OrderActivities activities = Workflow.newActivityStub(
        OrderActivities.class,
        ActivityOptions.newBuilder()
            .setStartToCloseTimeout(Duration.ofMinutes(5))
            .setRetryOptions(RetryOptions.newBuilder()
                .setMaximumAttempts(3)
                .build())
            .build()
    );

    private OrderStatus status = OrderStatus.CREATED;
    private final List<OrderItem> items = new ArrayList<>();
    private Address shippingAddress;

    @Override
    public OrderResult processOrder(OrderRequest order) {
        this.shippingAddress = order.getShippingAddress();
        this.items.addAll(order.getItems());

        // Validate order
        status = OrderStatus.VALIDATING;
        boolean isValid = activities.validateOrder(order);
        if (!isValid) {
            status = OrderStatus.REJECTED;
            return new OrderResult(false, "Order validation failed");
        }

        // Process payment
        status = OrderStatus.PROCESSING_PAYMENT;
        PaymentResult payment = activities.processPayment(order.getPaymentInfo());
        if (!payment.isSuccessful()) {
            status = OrderStatus.PAYMENT_FAILED;
            return new OrderResult(false, "Payment failed: " + payment.getErrorMessage());
        }

        // Create shipment
        status = OrderStatus.CREATING_SHIPMENT;
        ShipmentInfo shipment = activities.createShipment(shippingAddress, items);

        // Send confirmation
        activities.sendOrderConfirmation(order.getCustomerEmail(), shipment);

        status = OrderStatus.COMPLETED;
        return new OrderResult(true, shipment.getTrackingNumber());
    }

    @Override
    public void addItem(OrderItem item) {
        if (status == OrderStatus.CREATED) {
            items.add(item);
        }
    }

    @Override
    public OrderStatus getStatus() {
        return status;
    }

    @Override
    public void updateShippingAddress(Address newAddress) {
        if (status.ordinal() < OrderStatus.CREATING_SHIPMENT.ordinal()) {
            this.shippingAddress = newAddress;
        } else {
            throw new IllegalStateException("Cannot update address after shipment creation");
        }
    }
}

/**
 * Information about the current workflow execution.
 */
public interface WorkflowInfo {
    /**
     * Workflow execution namespace.
     * @return namespace name
     */
    String getNamespace();

    /**
     * Workflow ID.
     * @return workflow ID
     */
    String getWorkflowId();

    /**
     * Workflow run ID.
     * @return run ID
     */
    String getRunId();

    /**
     * Workflow type name.
     * @return workflow type
     */
    String getWorkflowType();

    /**
     * Task queue name.
     * @return task queue
     */
    String getTaskQueue();

    /**
     * Workflow execution start time.
     * @return start time in milliseconds since epoch
     */
    long getWorkflowStartTime();

    /**
     * Workflow execution timeout.
     * @return execution timeout duration
     */
    Duration getWorkflowExecutionTimeout();

    /**
     * Workflow run timeout.
     * @return run timeout duration
     */
    Duration getWorkflowRunTimeout();

    /**
     * Workflow task timeout.
     * @return task timeout duration
     */
    Duration getWorkflowTaskTimeout();

    /**
     * Workflow attempt number.
     * @return attempt number starting from 1
     */
    int getAttempt();

    /**
     * Cron schedule if workflow is scheduled.
     * @return optional cron schedule
     */
    Optional<String> getCronSchedule();

    /**
     * Continued execution run ID if this is a continued workflow.
     * @return optional continued run ID
     */
    Optional<String> getContinuedExecutionRunId();

    /**
     * Parent workflow execution if this is a child workflow.
     * @return optional parent execution
     */
    Optional<WorkflowExecution> getParentWorkflowExecution();

    /**
     * Parent namespace if this is a child workflow.
     * @return optional parent namespace
     */
    Optional<String> getParentNamespace();

    /**
     * Search attributes for this workflow.
     * @return search attributes map
     */
    Map<String, Object> getSearchAttributes();

    /**
     * Typed search attributes for this workflow.
     * @return typed search attributes
     */
    SearchAttributes getTypedSearchAttributes();

    /**
     * Memo attached to this workflow.
     * @return memo map
     */
    Map<String, Object> getMemo();
}

/**
 * Client stub for child workflow management.
 */
public interface ChildWorkflowStub {
    /**
     * Start child workflow execution.
     * @param args workflow arguments
     * @return promise that completes when workflow starts
     */
    Promise<WorkflowExecution> start(Object... args);

    /**
     * Execute child workflow and get result.
     * @param resultClass result class
     * @param args workflow arguments
     * @return promise with workflow result
     */
    <R> Promise<R> execute(Class<R> resultClass, Object... args);

    /**
     * Execute child workflow with generic type support.
     * @param resultClass result class
     * @param resultType generic type information
     * @param args workflow arguments
     * @return promise with workflow result
     */
    <R> Promise<R> execute(Class<R> resultClass, Type resultType, Object... args);

    /**
     * Get child workflow execution info.
     * @return workflow execution
     */
    WorkflowExecution getExecution();

    /**
     * Signal child workflow.
     * @param signalName signal name
     * @param args signal arguments
     */
    void signal(String signalName, Object... args);

    /**
     * Query child workflow.
     * @param queryType query type
     * @param resultClass result class
     * @param args query arguments
     * @return query result
     */
    <R> R query(String queryType, Class<R> resultClass, Object... args);
}

public class ParentWorkflowImpl implements ParentWorkflow {
    @Override
    public String orchestrateProcessing(ProcessingRequest request) {
        // Create child workflow stub
        DataProcessingWorkflow childWorkflow = Workflow.newChildWorkflowStub(
            DataProcessingWorkflow.class,
            ChildWorkflowOptions.newBuilder()
                .setWorkflowId("data-processing-" + request.getId())
                .setWorkflowExecutionTimeout(Duration.ofHours(1))
                .build()
        );

        // Start multiple child workflows
        List<Promise<String>> childResults = new ArrayList<>();
        for (DataChunk chunk : request.getChunks()) {
            Promise<String> result = childWorkflow.processDataChunk(chunk);
            childResults.add(result);
        }

        // Wait for all child workflows to complete
        Promise<Void> allDone = Promise.allOf(childResults);
        allDone.get();

        // Aggregate results
        StringBuilder aggregatedResult = new StringBuilder();
        for (Promise<String> result : childResults) {
            aggregatedResult.append(result.get()).append("\n");
        }

        return aggregatedResult.toString();
    }
}

/**
 * Client stub for signaling external workflows.
 */
public interface ExternalWorkflowStub {
    /**
     * Send signal to external workflow.
     * @param signalName signal name
     * @param args signal arguments
     */
    void signal(String signalName, Object... args);

    /**
     * Cancel external workflow.
     */
    void cancel();

    /**
     * Get external workflow execution info.
     * @return workflow execution
     */
    WorkflowExecution getExecution();
}

public class CoordinatorWorkflowImpl implements CoordinatorWorkflow {
    @Override
    public void coordinateProcessing(List<String> workerWorkflowIds) {
        // Signal all worker workflows to start
        List<ExternalWorkflowStub> workers = new ArrayList<>();
        for (String workflowId : workerWorkflowIds) {
            ExternalWorkflowStub worker = Workflow.newExternalWorkflowStub(
                WorkerWorkflow.class,
                workflowId
            );
            workers.add(worker);
            worker.signal("startProcessing");
        }

        // Wait for some condition
        Workflow.sleep(Duration.ofMinutes(30));

        // Signal all workers to stop
        for (ExternalWorkflowStub worker : workers) {
            worker.signal("stopProcessing");
        }
    }
}

/**
 * Interface for implementing workflows that handle multiple workflow types dynamically.
 */
public interface DynamicWorkflow {
    /**
     * Execute dynamic workflow.
     * @param args workflow arguments
     * @return workflow result
     */
    Object execute(EncodedValues args);
}

public class GenericWorkflowImpl implements DynamicWorkflow {
    @Override
    public Object execute(EncodedValues args) {
        WorkflowInfo info = Workflow.getInfo();
        String workflowType = info.getWorkflowType();

        switch (workflowType) {
            case "DataProcessing":
                return handleDataProcessing(args);
            case "OrderFulfillment":
                return handleOrderFulfillment(args);
            case "UserOnboarding":
                return handleUserOnboarding(args);
            default:
                throw new IllegalArgumentException("Unknown workflow type: " + workflowType);
        }
    }

    private Object handleDataProcessing(EncodedValues args) {
        String dataSource = args.get(0, String.class);
        ProcessingOptions options = args.get(1, ProcessingOptions.class);

        // Process data...
        return new ProcessingResult("Processed: " + dataSource);
    }
}

/**
 * Scope for managing cancellation of workflow operations.
 */
public interface CancellationScope {
    /**
     * Execute runnable in this cancellation scope.
     * @param runnable code to execute
     */
    void run(Runnable runnable);

    /**
     * Cancel this scope and all nested operations.
     */
    void cancel();

    /**
     * Cancel this scope with reason.
     * @param reason cancellation reason
     */
    void cancel(String reason);

    /**
     * Check if this scope is canceled.
     * @return true if canceled
     */
    boolean isCanceled();

    /**
     * Get cancellation reason if available.
     * @return optional cancellation reason
     */
    Optional<String> getCancellationReason();
}

public class TimeoutWorkflowImpl implements TimeoutWorkflow {
    @Override
    public String processWithTimeout(ProcessingRequest request) {
        CancellationScope timeoutScope = Workflow.newCancellationScope(false, () -> {
            // Long running operation
            ProcessingActivity activity = Workflow.newActivityStub(ProcessingActivity.class);
            return activity.processData(request);
        });

        // Set up timeout
        Workflow.newTimer(Duration.ofMinutes(30)).thenRun(() -> {
            timeoutScope.cancel("Processing timeout");
        });

        try {
            timeoutScope.run(() -> {
                // This will be canceled if timeout occurs
            });
            return "Processing completed";
        } catch (CanceledFailure e) {
            return "Processing timed out";
        }
    }
}

/**
 * Marks interface as a workflow interface.
 */
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
public @interface WorkflowInterface {
}

/**
 * Marks the main workflow method.
 */
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface WorkflowMethod {
    /**
     * Workflow type name.
     * @return workflow type (default: method name with first letter capitalized)
     */
    String name() default "";
}

/**
 * Marks workflow signal handler methods.
 */
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface SignalMethod {
    /**
     * Signal name.
     * @return signal name (default: method name)
     */
    String name() default "";
}

/**
 * Marks workflow query handler methods.
 */
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface QueryMethod {
    /**
     * Query name.
     * @return query name (default: method name)
     */
    String name() default "";
}

/**
 * Marks workflow update handler methods.
 */
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface UpdateMethod {
    /**
     * Update name.
     * @return update name (default: method name)
     */
    String name() default "";
}

/**
 * Configuration for child workflow execution.
 */
public final class ChildWorkflowOptions {
    /**
     * Creates new builder.
     * @return new ChildWorkflowOptions builder
     */
    public static Builder newBuilder();

    /**
     * Creates builder from existing options.
     * @param options existing options to copy
     * @return new builder with copied options
     */
    public static Builder newBuilder(ChildWorkflowOptions options);

    /**
     * Builder for ChildWorkflowOptions.
     */
    public static final class Builder {
        /**
         * Child workflow ID.
         * @param workflowId workflow ID
         * @return this builder
         */
        public Builder setWorkflowId(String workflowId);

        /**
         * Workflow execution timeout.
         * @param timeout execution timeout
         * @return this builder
         */
        public Builder setWorkflowExecutionTimeout(Duration timeout);

        /**
         * Workflow run timeout.
         * @param timeout run timeout
         * @return this builder
         */
        public Builder setWorkflowRunTimeout(Duration timeout);

        /**
         * Workflow task timeout.
         * @param timeout task timeout
         * @return this builder
         */
        public Builder setWorkflowTaskTimeout(Duration timeout);

        /**
         * Task queue for child workflow.
         * @param taskQueue task queue name
         * @return this builder
         */
        public Builder setTaskQueue(String taskQueue);

        /**
         * Retry options for child workflow.
         * @param retryOptions retry options
         * @return this builder
         */
        public Builder setRetryOptions(RetryOptions retryOptions);

        /**
         * Cron schedule for child workflow.
         * @param cronSchedule cron expression
         * @return this builder
         */
        public Builder setCronSchedule(String cronSchedule);

        /**
         * Workflow ID reuse policy.
         * @param workflowIdReusePolicy reuse policy
         * @return this builder
         */
        public Builder setWorkflowIdReusePolicy(WorkflowIdReusePolicy workflowIdReusePolicy);

        /**
         * Parent close policy.
         * @param parentClosePolicy parent close policy
         * @return this builder
         */
        public Builder setParentClosePolicy(ParentClosePolicy parentClosePolicy);

        /**
         * Memo for child workflow.
         * @param memo memo map
         * @return this builder
         */
        public Builder setMemo(Map<String, Object> memo);

        /**
         * Search attributes for child workflow.
         * @param searchAttributes search attributes map
         * @return this builder
         */
        public Builder setSearchAttributes(Map<String, ?> searchAttributes);

        /**
         * Typed search attributes for child workflow.
         * @param searchAttributes typed search attributes
         * @return this builder
         */
        public Builder setTypedSearchAttributes(SearchAttributes searchAttributes);

        /**
         * Build the ChildWorkflowOptions.
         * @return configured ChildWorkflowOptions
         */
        public ChildWorkflowOptions build();
    }
}

/**
 * Options for continue-as-new functionality.
 */
public final class ContinueAsNewOptions {
    /**
     * Creates new builder.
     * @return new ContinueAsNewOptions builder
     */
    public static Builder newBuilder();

    /**
     * Builder for ContinueAsNewOptions.
     */
    public static final class Builder {
        /**
         * Workflow type for the new execution.
         * @param workflowType workflow type name
         * @return this builder
         */
        public Builder setWorkflowType(String workflowType);

        /**
         * Task queue for the new execution.
         * @param taskQueue task queue name
         * @return this builder
         */
        public Builder setTaskQueue(String taskQueue);

        /**
         * Workflow run timeout for the new execution.
         * @param timeout run timeout
         * @return this builder
         */
        public Builder setWorkflowRunTimeout(Duration timeout);

        /**
         * Workflow task timeout for the new execution.
         * @param timeout task timeout
         * @return this builder
         */
        public Builder setWorkflowTaskTimeout(Duration timeout);

        /**
         * Memo for the new execution.
         * @param memo memo map
         * @return this builder
         */
        public Builder setMemo(Map<String, Object> memo);

        /**
         * Search attributes for the new execution.
         * @param searchAttributes search attributes map
         * @return this builder
         */
        public Builder setSearchAttributes(Map<String, ?> searchAttributes);

        /**
         * Typed search attributes for the new execution.
         * @param searchAttributes typed search attributes
         * @return this builder
         */
        public Builder setTypedSearchAttributes(SearchAttributes searchAttributes);

        /**
         * Build the ContinueAsNewOptions.
         * @return configured ContinueAsNewOptions
         */
        public ContinueAsNewOptions build();
    }
}

public class ProcessingWorkflowImpl implements ProcessingWorkflow {
    @Override
    public void processInBatches(BatchRequest request) {
        List<DataItem> currentBatch = request.getBatch();

        // Process current batch
        for (DataItem item : currentBatch) {
            processItem(item);
        }

        // Check if more batches to process
        if (request.hasMoreBatches()) {
            BatchRequest nextRequest = request.getNextBatch();

            // Continue as new to avoid history growth
            Workflow.continueAsNew(
                ContinueAsNewOptions.newBuilder()
                    .setMemo(Map.of("processed_batches", request.getProcessedBatchCount() + 1))
                    .build(),
                nextRequest
            );
        }
    }
}