/*
 * Copyright 2022 Conductor Authors.
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
 * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations under the License.
 */
package com.netflix.conductor.core.execution;

import java.time.Duration;
import java.util.*;
import java.util.function.Predicate;
import java.util.stream.Collectors;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.StopWatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import com.netflix.conductor.annotations.Trace;
import com.netflix.conductor.annotations.VisibleForTesting;
import com.netflix.conductor.common.metadata.tasks.*;
import com.netflix.conductor.common.metadata.workflow.RerunWorkflowRequest;
import com.netflix.conductor.common.metadata.workflow.SkipTaskRequest;
import com.netflix.conductor.common.metadata.workflow.WorkflowDef;
import com.netflix.conductor.common.metadata.workflow.WorkflowTask;
import com.netflix.conductor.common.run.Workflow;
import com.netflix.conductor.common.utils.TaskUtils;
import com.netflix.conductor.core.WorkflowContext;
import com.netflix.conductor.core.config.ConductorProperties;
import com.netflix.conductor.core.dal.ExecutionDAOFacade;
import com.netflix.conductor.core.exception.*;
import com.netflix.conductor.core.execution.tasks.SystemTaskRegistry;
import com.netflix.conductor.core.execution.tasks.Terminate;
import com.netflix.conductor.core.execution.tasks.WorkflowSystemTask;
import com.netflix.conductor.core.listener.TaskStatusListener;
import com.netflix.conductor.core.listener.WorkflowStatusListener;
import com.netflix.conductor.core.listener.WorkflowStatusListener.WorkflowEventType;
import com.netflix.conductor.core.metadata.MetadataMapperService;
import com.netflix.conductor.core.utils.IDGenerator;
import com.netflix.conductor.core.utils.ParametersUtils;
import com.netflix.conductor.core.utils.QueueUtils;
import com.netflix.conductor.core.utils.Utils;
import com.netflix.conductor.dao.MetadataDAO;
import com.netflix.conductor.dao.QueueDAO;
import com.netflix.conductor.dao.WorkflowMessageQueueDAO;
import com.netflix.conductor.metrics.Monitors;
import com.netflix.conductor.model.TaskModel;
import com.netflix.conductor.model.WorkflowModel;
import com.netflix.conductor.service.ExecutionLockService;

import com.google.common.base.Preconditions;

import static com.netflix.conductor.core.utils.Utils.DECIDER_QUEUE;
import static com.netflix.conductor.model.TaskModel.Status.*;

import static org.conductoross.conductor.core.execution.ExecutorUtils.computePostpone;
import static org.conductoross.conductor.core.execution.ExecutorUtils.hasInProgressHumanTask;

/** Workflow services provider interface */
@Trace
@Component
public class WorkflowExecutorOps implements WorkflowExecutor {

    private static final Logger LOGGER = LoggerFactory.getLogger(WorkflowExecutorOps.class);
    private static final int EXPEDITED_PRIORITY = 10;
    private static final String CLASS_NAME = WorkflowExecutor.class.getSimpleName();
    private static final Predicate<TaskModel> UNSUCCESSFUL_TERMINAL_TASK =
            task -> !task.getStatus().isSuccessful() && task.getStatus().isTerminal();
    private static final Predicate<TaskModel> UNSUCCESSFUL_JOIN_TASK =
            UNSUCCESSFUL_TERMINAL_TASK.and(t -> TaskType.TASK_TYPE_JOIN.equals(t.getTaskType()));
    private static final Predicate<TaskModel> NON_TERMINAL_TASK =
            task -> !task.getStatus().isTerminal();
    private final MetadataDAO metadataDAO;
    private final QueueDAO queueDAO;
    private final DeciderService deciderService;
    private final ConductorProperties properties;
    private final MetadataMapperService metadataMapperService;
    private final ExecutionDAOFacade executionDAOFacade;
    private final ParametersUtils parametersUtils;
    private final IDGenerator idGenerator;
    private final WorkflowStatusListener workflowStatusListener;
    private final TaskStatusListener taskStatusListener;
    private final SystemTaskRegistry systemTaskRegistry;
    private long activeWorkerLastPollMs;
    private final ExecutionLockService executionLockService;
    private final Optional<WorkflowMessageQueueDAO> workflowMessageQueueDAO;

    private final Predicate<PollData> validateLastPolledTime =
            pollData ->
                    pollData.getLastPollTime()
                            > System.currentTimeMillis() - activeWorkerLastPollMs;

    public WorkflowExecutorOps(
            DeciderService deciderService,
            MetadataDAO metadataDAO,
            QueueDAO queueDAO,
            MetadataMapperService metadataMapperService,
            WorkflowStatusListener workflowStatusListener,
            TaskStatusListener taskStatusListener,
            ExecutionDAOFacade executionDAOFacade,
            ConductorProperties properties,
            ExecutionLockService executionLockService,
            SystemTaskRegistry systemTaskRegistry,
            ParametersUtils parametersUtils,
            IDGenerator idGenerator,
            Optional<WorkflowMessageQueueDAO> workflowMessageQueueDAO) {
        this.deciderService = deciderService;
        this.metadataDAO = metadataDAO;
        this.queueDAO = queueDAO;
        this.properties = properties;
        this.metadataMapperService = metadataMapperService;
        this.executionDAOFacade = executionDAOFacade;
        this.activeWorkerLastPollMs = properties.getActiveWorkerLastPollTimeout().toMillis();
        this.workflowStatusListener = workflowStatusListener;
        this.taskStatusListener = taskStatusListener;
        this.executionLockService = executionLockService;
        this.parametersUtils = parametersUtils;
        this.idGenerator = idGenerator;
        this.systemTaskRegistry = systemTaskRegistry;
        this.workflowMessageQueueDAO = workflowMessageQueueDAO;
    }

    /**
     * @param workflowId the id of the workflow for which task callbacks are to be reset
     * @throws ConflictException if the workflow is in terminal state
     */
    @Override
    public void resetCallbacksForWorkflow(String workflowId) {
        WorkflowModel workflow = executionDAOFacade.getWorkflowModel(workflowId, true);
        if (workflow.getStatus().isTerminal()) {
            throw new ConflictException(
                    "Workflow is in terminal state. Status = %s", workflow.getStatus());
        }

        // Get SIMPLE tasks in SCHEDULED state that have callbackAfterSeconds > 0 and
        // set the
        // callbackAfterSeconds to 0
        workflow.getTasks().stream()
                .filter(
                        task ->
                                !systemTaskRegistry.isSystemTask(task.getTaskType())
                                        && SCHEDULED == task.getStatus()
                                        && task.getCallbackAfterSeconds() > 0)
                .forEach(
                        task -> {
                            if (queueDAO.resetOffsetTime(
                                    QueueUtils.getQueueName(task), task.getTaskId())) {
                                task.setCallbackAfterSeconds(0);
                                executionDAOFacade.updateTask(task);
                            }
                        });
    }

    @Override
    public String rerun(RerunWorkflowRequest request) {
        Utils.checkNotNull(request.getReRunFromWorkflowId(), "reRunFromWorkflowId is missing");
        if (!rerunWF(
                request.getReRunFromWorkflowId(),
                request.getReRunFromTaskId(),
                request.getTaskInput(),
                request.getWorkflowInput(),
                request.getCorrelationId())) {
            throw new IllegalArgumentException(
                    "Task " + request.getReRunFromTaskId() + " not found");
        }
        return request.getReRunFromWorkflowId();
    }

    /**
     * @param workflowId the id of the workflow to be restarted
     * @param useLatestDefinitions if true, use the latest workflow and task definitions upon
     *     restart
     * @throws ConflictException Workflow is not in a terminal state.
     * @throws NotFoundException Workflow definition is not found or Workflow is deemed
     *     non-restartable as per workflow definition.
     */
    @Override
    public void restart(String workflowId, boolean useLatestDefinitions) {
        final WorkflowModel workflow = executionDAOFacade.getWorkflowModel(workflowId, true);

        if (!workflow.getStatus().isTerminal()) {
            String errorMsg =
                    String.format(
                            "Workflow: %s is not in terminal state, unable to restart.", workflow);
            LOGGER.error(errorMsg);
            throw new ConflictException(errorMsg);
        }

        WorkflowDef workflowDef;
        if (useLatestDefinitions) {
            workflowDef =
                    metadataDAO
                            .getLatestWorkflowDef(workflow.getWorkflowName())
                            .orElseThrow(
                                    () ->
                                            new NotFoundException(
                                                    "Unable to find latest definition for %s",
                                                    workflowId));
            workflow.setWorkflowDefinition(workflowDef);
            workflowDef = metadataMapperService.populateTaskDefinitions(workflowDef);
        } else {
            workflowDef =
                    Optional.ofNullable(workflow.getWorkflowDefinition())
                            .orElseGet(
                                    () ->
                                            metadataDAO
                                                    .getWorkflowDef(
                                                            workflow.getWorkflowName(),
                                                            workflow.getWorkflowVersion())
                                                    .orElseThrow(
                                                            () ->
                                                                    new NotFoundException(
                                                                            "Unable to find definition for %s",
                                                                            workflowId)));
        }

        if (!workflowDef.isRestartable()
                && workflow.getStatus()
                        .equals(
                                WorkflowModel.Status
                                        .COMPLETED)) { // Can only restart non-completed workflows
            // when the configuration is set to false
            throw new NotFoundException("Workflow: %s is non-restartable", workflow);
        }

        // Reset the workflow in the primary datastore and remove from indexer; then
        // re-create it
        executionDAOFacade.resetWorkflow(workflowId);

        workflow.getTasks().clear();
        workflow.setReasonForIncompletion(null);
        workflow.setFailedTaskId(null);
        workflow.setCreateTime(System.currentTimeMillis());
        workflow.setEndTime(0);
        workflow.setLastRetriedTime(0);
        // Change the status to running
        workflow.setStatus(WorkflowModel.Status.RUNNING);
        workflow.setOutput(null);
        workflow.setExternalOutputPayloadStoragePath(null);

        try {
            executionDAOFacade.createWorkflow(workflow);
            // Notify on workflow started.
            notifyWorkflowStatusListener(workflow, WorkflowEventType.RESTARTED);
        } catch (Exception e) {
            Monitors.recordWorkflowStartError(
                    workflowDef.getName(), WorkflowContext.get().getClientApp());
            LOGGER.error("Unable to restart workflow: {}", workflowDef.getName(), e);
            terminateWorkflow(workflowId, "Error when restarting the workflow");
            throw e;
        }

        metadataMapperService.populateWorkflowWithDefinitions(workflow);
        decide(workflowId);

        updateAndPushParents(workflow, "restarted");
    }

    /**
     * Gets the last instance of each failed task and reschedule each Gets all cancelled tasks and
     * schedule all of them except JOIN (join should change status to INPROGRESS) Switch workflow
     * back to RUNNING status and call decider.
     *
     * @param workflowId the id of the workflow to be retried
     */
    @Override
    public void retry(String workflowId, boolean resumeSubworkflowTasks) {
        WorkflowModel workflow = executionDAOFacade.getWorkflowModel(workflowId, true);
        if (!workflow.getStatus().isTerminal()) {
            throw new NotFoundException(
                    "Workflow is still running.  status=%s", workflow.getStatus());
        }
        if (workflow.getTasks().isEmpty()) {
            throw new ConflictException("Workflow has not started yet");
        }

        if (resumeSubworkflowTasks) {
            Optional<TaskModel> taskToRetry =
                    workflow.getTasks().stream().filter(UNSUCCESSFUL_TERMINAL_TASK).findFirst();
            if (taskToRetry.isPresent()) {
                workflow = findLastFailedSubWorkflowIfAny(taskToRetry.get(), workflow);
                retry(workflow);
                updateAndPushParents(workflow, "retried");
                decide(workflow.getWorkflowId());
            }
        } else {
            retry(workflow);
            updateAndPushParents(workflow, "retried");
            decide(workflow.getWorkflowId());
        }
    }

    private void updateAndPushParents(WorkflowModel workflow, String operation) {
        String workflowIdentifier = "";
        while (workflow.hasParent()) {
            // update parent's sub workflow task
            TaskModel subWorkflowTask =
                    executionDAOFacade.getTaskModel(workflow.getParentWorkflowTaskId());
            if (subWorkflowTask == null || subWorkflowTask.getWorkflowTask() == null) {
                // orphan sub-workflow: parent task reference no longer exists (e.g. parent was
                // restarted and task list was cleared) — stop walking parent chain
                break;
            }
            if (subWorkflowTask.getWorkflowTask().isOptional()) {
                LOGGER.info(
                        "Sub workflow task {} is optional, skip updating parents", subWorkflowTask);
                break;
            }
            if (subWorkflowTask.isRetried()
                    && TaskType.TASK_TYPE_SUB_WORKFLOW.equalsIgnoreCase(
                            subWorkflowTask.getTaskType())) {
                // this sub-workflow belongs to a superseded retry attempt; the parent has already
                // advanced to a newer task — stop walking
                break;
            }
            subWorkflowTask.setSubworkflowChanged(true);
            subWorkflowTask.setStatus(IN_PROGRESS);
            subWorkflowTask.setReasonForIncompletion(null);
            executionDAOFacade.updateTask(subWorkflowTask);

            // add an execution log
            String currentWorkflowIdentifier = workflow.toShortString();
            workflowIdentifier =
                    !workflowIdentifier.equals("")
                            ? String.format(
                                    "%s -> %s", currentWorkflowIdentifier, workflowIdentifier)
                            : currentWorkflowIdentifier;
            TaskExecLog log =
                    new TaskExecLog(
                            String.format("Sub workflow %s %s.", workflowIdentifier, operation));
            log.setTaskId(subWorkflowTask.getTaskId());
            executionDAOFacade.addTaskExecLog(Collections.singletonList(log));
            LOGGER.info("Task {} updated. {}", log.getTaskId(), log.getLog());

            // push the parent workflow to decider queue for asynchronous 'decide'
            String parentWorkflowId = workflow.getParentWorkflowId();
            WorkflowModel parentWorkflow =
                    executionDAOFacade.getWorkflowModel(parentWorkflowId, true);
            parentWorkflow.setStatus(WorkflowModel.Status.RUNNING);
            parentWorkflow.setReasonForIncompletion(null);
            parentWorkflow.setFailedTaskId(null);
            parentWorkflow.setFailedReferenceTaskNames(new HashSet<>());
            parentWorkflow.setFailedTaskNames(new HashSet<>());
            parentWorkflow.setLastRetriedTime(System.currentTimeMillis());
            executionDAOFacade.updateWorkflow(parentWorkflow);

            for (TaskModel task : parentWorkflow.getTasks()) {
                if (task.getTaskType().equalsIgnoreCase(TaskType.TASK_TYPE_SUB_WORKFLOW)
                        && task.getSubWorkflowId() != null
                        && UNSUCCESSFUL_TERMINAL_TASK.test(task)) {
                    // retry sibling sub-workflows that are still in a failed/timed-out state
                    WorkflowModel child =
                            executionDAOFacade.getWorkflowModel(task.getSubWorkflowId(), true);
                    if (child != null) {
                        if (child.getStatus() == WorkflowModel.Status.RUNNING) {
                            // Child was already set RUNNING by an in-progress rerun; surfacing that
                            // to the parent task without calling retry() avoids creating a spurious
                            // new task instance that conflicts with the rerun's own finalizeRerun.
                            task.setStatus(IN_PROGRESS);
                            task.setReasonForIncompletion(null);
                            task.setSubworkflowChanged(true);
                            executionDAOFacade.updateTask(task);
                        } else if (child.getTasks().stream().anyMatch(UNSUCCESSFUL_TERMINAL_TASK)) {
                            retry(child);
                            task.setStatus(IN_PROGRESS);
                            task.setReasonForIncompletion(null);
                            task.setSubworkflowChanged(true);
                            executionDAOFacade.updateTask(task);
                        }
                    }
                } else if (task.getStatus() == CANCELED) {
                    if (task.getTaskType().equalsIgnoreCase(TaskType.JOIN.toString())
                            || task.getTaskType().equalsIgnoreCase(TaskType.DO_WHILE.toString())) {
                        task.setStatus(IN_PROGRESS);
                        executionDAOFacade.updateTask(task);
                    } else {
                        task.setRetryCount(task.getRetryCount() + 1);
                        task.setReasonForIncompletion(null);
                        task.setPollCount(0);
                        task.setWorkerId(null);
                        task.setScheduledTime(System.currentTimeMillis());
                        task.setStartTime(0);
                        task.setEndTime(0);
                        task.setRetried(false);
                        task.setExecuted(false);
                        task.setStatus(SCHEDULED);
                        executionDAOFacade.updateTask(task);
                        addTaskToQueue(task);
                    }
                }
            }

            try {
                WorkflowStatusListener.WorkflowEventType event =
                        WorkflowStatusListener.WorkflowEventType.valueOf(operation.toUpperCase());
                notifyWorkflowStatusListener(parentWorkflow, event);
            } catch (IllegalArgumentException e) {
                LOGGER.warn("Unknown workflow operation: {}", operation);
            }

            expediteLazyWorkflowEvaluation(parentWorkflowId);

            workflow = parentWorkflow;
        }
    }

    private void resetUnsuccessfulJoinTasks(WorkflowModel workflow) {
        if (workflow.getWorkflowDefinition().containsType(TaskType.TASK_TYPE_JOIN)
                || workflow.getWorkflowDefinition()
                        .containsType(TaskType.TASK_TYPE_FORK_JOIN_DYNAMIC)) {
            workflow.getTasks().stream()
                    .filter(UNSUCCESSFUL_JOIN_TASK)
                    .peek(
                            task -> {
                                task.setStatus(TaskModel.Status.IN_PROGRESS);
                                addTaskToQueue(task);
                            })
                    .forEach(executionDAOFacade::updateTask);
        }
    }

    private void retry(WorkflowModel workflow) {
        // Get all FAILED or CANCELED tasks that are not COMPLETED (or reach other
        // terminal states)
        // on further executions.
        // // Eg: for Seq of tasks task1.CANCELED, task1.COMPLETED, task1 shouldn't be
        // retried.
        // Throw an exception if there are no FAILED tasks.
        // Handle JOIN task CANCELED status as special case.
        Map<String, TaskModel> retriableMap = new HashMap<>();
        for (TaskModel task : workflow.getTasks()) {
            switch (task.getStatus()) {
                case FAILED:
                    if (task.getTaskType().equalsIgnoreCase(TaskType.JOIN.toString())
                            || task.getTaskType()
                                    .equalsIgnoreCase(TaskType.EXCLUSIVE_JOIN.toString())) {
                        @SuppressWarnings("unchecked")
                        List<String> joinOn = (List<String>) task.getInputData().get("joinOn");
                        boolean joinOnFailedPermissive = isJoinOnFailedPermissive(joinOn, workflow);
                        if (joinOnFailedPermissive) {
                            task.setStatus(IN_PROGRESS);
                            addTaskToQueue(task);
                            break;
                        }
                    }
                case FAILED_WITH_TERMINAL_ERROR:
                case TIMED_OUT:
                    retriableMap.put(task.getReferenceTaskName(), task);
                    break;
                case CANCELED:
                    if (task.getTaskType().equalsIgnoreCase(TaskType.JOIN.toString())
                            || task.getTaskType().equalsIgnoreCase(TaskType.DO_WHILE.toString())) {
                        task.setStatus(IN_PROGRESS);
                        addTaskToQueue(task);
                        // Task doesn't have to be updated yet. Will be updated along with other
                        // Workflow tasks downstream.
                    } else {
                        retriableMap.put(task.getReferenceTaskName(), task);
                    }
                    break;
                default:
                    retriableMap.remove(task.getReferenceTaskName());
                    break;
            }
        }

        // if workflow TIMED_OUT due to timeoutSeconds configured in the workflow
        // definition,
        // it may not have any unsuccessful tasks that can be retried
        if (retriableMap.values().size() == 0
                && workflow.getStatus() != WorkflowModel.Status.TIMED_OUT) {
            throw new ConflictException(
                    "There are no retryable tasks! Use restart if you want to attempt entire workflow execution again.");
        }

        // Update Workflow with new status.
        // This should load Workflow from archive, if archived.
        workflow.setStatus(WorkflowModel.Status.RUNNING);
        workflow.setLastRetriedTime(System.currentTimeMillis());
        String lastReasonForIncompletion = workflow.getReasonForIncompletion();
        workflow.setReasonForIncompletion(null);
        executionDAOFacade.updateWorkflow(workflow);
        notifyWorkflowStatusListener(workflow, WorkflowEventType.RETRIED);
        LOGGER.info(
                "Workflow {} that failed due to '{}' was retried",
                workflow.toShortString(),
                lastReasonForIncompletion);

        // taskToBeRescheduled would set task `retried` to true, and hence it's
        // important to
        // updateTasks after obtaining task copy from taskToBeRescheduled.
        final WorkflowModel finalWorkflow = workflow;
        List<TaskModel> retriableTasks =
                retriableMap.values().stream()
                        .sorted(Comparator.comparingInt(TaskModel::getSeq))
                        .map(task -> taskToBeRescheduled(finalWorkflow, task))
                        .collect(Collectors.toList());

        dedupAndAddTasks(workflow, retriableTasks);
        // Note: updateTasks before updateWorkflow might fail when Workflow is archived
        // and doesn't
        // exist in primary store.
        executionDAOFacade.updateTasks(workflow.getTasks());
        scheduleTask(workflow, retriableTasks);
        // Push AFTER tasks are reset so async decider sees SCHEDULED/IN_PROGRESS, not stale state
        queueDAO.push(
                DECIDER_QUEUE,
                workflow.getWorkflowId(),
                workflow.getPriority(),
                properties.getWorkflowOffsetTimeout().getSeconds());
    }

    private WorkflowModel findLastFailedSubWorkflowIfAny(
            TaskModel task, WorkflowModel parentWorkflow) {
        if (!UNSUCCESSFUL_TERMINAL_TASK.test(task)) return parentWorkflow;

        if (TaskType.TASK_TYPE_SUB_WORKFLOW.equals(task.getTaskType())) {
            WorkflowModel subWorkflow =
                    executionDAOFacade.getWorkflowModel(task.getSubWorkflowId(), true);
            return subWorkflow.getTasks().stream()
                    .filter(UNSUCCESSFUL_TERMINAL_TASK)
                    .findFirst()
                    .map(t -> findLastFailedSubWorkflowIfAny(t, subWorkflow))
                    .orElse(parentWorkflow);
        }

        if (TaskType.TASK_TYPE_DO_WHILE.equals(task.getTaskType())) {
            return parentWorkflow.getTasks().stream()
                    .filter(t -> TaskType.TASK_TYPE_SUB_WORKFLOW.equals(t.getTaskType()))
                    .filter(UNSUCCESSFUL_TERMINAL_TASK)
                    .findFirst()
                    .map(t -> findLastFailedSubWorkflowIfAny(t, parentWorkflow))
                    .orElse(parentWorkflow);
        }

        return parentWorkflow;
    }

    /**
     * Reschedule a task
     *
     * @param task failed or cancelled task
     * @return new instance of a task with "SCHEDULED" status
     */
    private TaskModel taskToBeRescheduled(WorkflowModel workflow, TaskModel task) {
        TaskModel taskToBeRetried = task.copy();
        taskToBeRetried.setTaskId(idGenerator.generate());
        taskToBeRetried.setRetriedTaskId(task.getTaskId());
        taskToBeRetried.setStatus(SCHEDULED);
        taskToBeRetried.setRetryCount(task.getRetryCount() + 1);
        taskToBeRetried.setRetried(false);
        taskToBeRetried.setPollCount(0);
        taskToBeRetried.setCallbackAfterSeconds(0);
        taskToBeRetried.setSubWorkflowId(null);
        taskToBeRetried.setScheduledTime(0);
        taskToBeRetried.setStartTime(0);
        taskToBeRetried.setEndTime(0);
        taskToBeRetried.setWorkerId(null);
        taskToBeRetried.setReasonForIncompletion(null);
        taskToBeRetried.setSeq(0);
        clearRetriedTaskRuntimeState(taskToBeRetried);

        // perform parameter replacement for retried task
        if (taskToBeRetried.getWorkflowTask() != null) {
            Map<String, Object> taskInput =
                    parametersUtils.getTaskInput(
                            taskToBeRetried.getWorkflowTask().getInputParameters(),
                            workflow,
                            taskToBeRetried.getWorkflowTask().getTaskDefinition(),
                            taskToBeRetried.getTaskId());
            taskToBeRetried.getInputData().putAll(taskInput);
        }
        clearLegacySubWorkflowId(taskToBeRetried);

        task.setRetried(true);
        // since this task is being retried and a retry has been computed, task
        // lifecycle is
        // complete
        task.setExecuted(true);
        return taskToBeRetried;
    }

    private void clearRetriedTaskRuntimeState(TaskModel task) {
        task.setUpdateTime(0);
        task.setCallbackAfterMs(0);
        task.setOutputData(new HashMap<>());
        task.setExternalOutputPayloadStoragePath(null);
        task.setOutputMessage(null);
    }

    private void clearLegacySubWorkflowId(TaskModel task) {
        task.setSubWorkflowId(null);
        task.getInputData().remove("subWorkflowId");
        task.getOutputData().remove("subWorkflowId");
    }

    private void endExecution(WorkflowModel workflow, TaskModel terminateTask) {
        boolean raiseFinalizedNotification = false;
        if (terminateTask != null) {
            String terminationStatus =
                    (String)
                            terminateTask
                                    .getInputData()
                                    .get(Terminate.getTerminationStatusParameter());
            String reason =
                    (String)
                            terminateTask
                                    .getInputData()
                                    .get(Terminate.getTerminationReasonParameter());
            if (StringUtils.isBlank(reason)) {
                reason =
                        String.format(
                                "Workflow is %s by TERMINATE task: %s",
                                terminationStatus, terminateTask.getTaskId());
            }
            if (WorkflowModel.Status.FAILED.name().equals(terminationStatus)) {
                workflow.setStatus(WorkflowModel.Status.FAILED);
                workflow =
                        terminate(
                                workflow,
                                new TerminateWorkflowException(
                                        reason, workflow.getStatus(), terminateTask));
            } else if (WorkflowModel.Status.TERMINATED.name().equals(terminationStatus)) {
                workflow.setStatus(WorkflowModel.Status.TERMINATED);
                workflow =
                        terminate(
                                workflow,
                                new TerminateWorkflowException(
                                        reason, workflow.getStatus(), terminateTask));
            } else {
                workflow.setReasonForIncompletion(reason);
                workflow = completeWorkflow(workflow);
                raiseFinalizedNotification = true;
            }
        } else {
            workflow = completeWorkflow(workflow);
            raiseFinalizedNotification = true;
        }
        cancelNonTerminalTasks(workflow, raiseFinalizedNotification);
    }

    /**
     * @param workflow the workflow to be completed
     * @throws ConflictException if workflow is already in terminal state.
     */
    @VisibleForTesting
    WorkflowModel completeWorkflow(WorkflowModel workflow) {
        LOGGER.debug("Completing workflow execution for {}", workflow.getWorkflowId());

        if (workflow.getStatus().equals(WorkflowModel.Status.COMPLETED)) {
            queueDAO.remove(DECIDER_QUEUE, workflow.getWorkflowId()); // remove from the sweep queue
            executionDAOFacade.removeFromPendingWorkflow(
                    workflow.getWorkflowName(), workflow.getWorkflowId());
            LOGGER.debug("Workflow: {} has already been completed.", workflow.getWorkflowId());
            return workflow;
        }

        if (workflow.getStatus().isTerminal()) {
            String msg =
                    "Workflow is already in terminal state. Current status: "
                            + workflow.getStatus();
            throw new ConflictException(msg);
        }

        deciderService.updateWorkflowOutput(workflow, null);

        workflow.setStatus(WorkflowModel.Status.COMPLETED);

        // update the failed reference task names
        List<TaskModel> failedTasks =
                workflow.getTasks().stream()
                        .filter(
                                t ->
                                        FAILED.equals(t.getStatus())
                                                || FAILED_WITH_TERMINAL_ERROR.equals(t.getStatus()))
                        .collect(Collectors.toList());

        workflow.getFailedReferenceTaskNames()
                .addAll(
                        failedTasks.stream()
                                .map(TaskModel::getReferenceTaskName)
                                .collect(Collectors.toSet()));

        workflow.getFailedTaskNames()
                .addAll(
                        failedTasks.stream()
                                .map(TaskModel::getTaskDefName)
                                .collect(Collectors.toSet()));

        executionDAOFacade.updateWorkflow(workflow);
        LOGGER.debug("Completed workflow execution for {}", workflow.getWorkflowId());
        notifyWorkflowStatusListener(workflow, WorkflowEventType.COMPLETED);
        Monitors.recordWorkflowCompletion(
                workflow.getWorkflowName(),
                workflow.getEndTime() - workflow.getCreateTime(),
                workflow.getOwnerApp());

        if (workflow.hasParent()) {
            updateParentWorkflowTask(workflow);
            LOGGER.info(
                    "{} updated parent {} task {}",
                    workflow.toShortString(),
                    workflow.getParentWorkflowId(),
                    workflow.getParentWorkflowTaskId());
            expediteLazyWorkflowEvaluation(workflow.getParentWorkflowId());
        }

        workflowMessageQueueDAO.ifPresent(dao -> dao.delete(workflow.getWorkflowId()));
        executionLockService.releaseLock(workflow.getWorkflowId());
        executionLockService.deleteLock(workflow.getWorkflowId());
        return workflow;
    }

    @Override
    public void terminateWorkflow(String workflowId, String reason) {
        WorkflowModel workflow = executionDAOFacade.getWorkflowModel(workflowId, true);
        if (WorkflowModel.Status.COMPLETED.equals(workflow.getStatus())) {
            throw new ConflictException("Cannot terminate a COMPLETED workflow.");
        }
        if (WorkflowModel.Status.TERMINATED.equals(workflow.getStatus())) {
            // Workflow is already in TERMINATED state; no additional termination action is
            // required.
            return;
        }
        workflow.setStatus(WorkflowModel.Status.TERMINATED);
        terminateWorkflow(workflow, reason, null);
    }

    /**
     * @param workflow Workflow to be terminated
     * @param reason Reason for termination
     * @param failureWorkflow Failure workflow (if any), to be triggered as a result of this
     *     termination
     * @param failureWorkflowVersion Failure workflow version (if any)
     */
    @Override
    public WorkflowModel terminateWorkflow(
            WorkflowModel workflow,
            String reason,
            String failureWorkflow,
            Integer failureWorkflowVersion) {
        try {
            executionLockService.acquireLock(workflow.getWorkflowId(), 60000);

            if (!workflow.getStatus().isTerminal()) {
                workflow.setStatus(WorkflowModel.Status.TERMINATED);
            }

            try {
                deciderService.updateWorkflowOutput(workflow, null);
            } catch (Exception e) {
                // catch any failure in this step and continue the execution of terminating
                // workflow
                LOGGER.error(
                        "Failed to update output data for workflow: {}",
                        workflow.getWorkflowId(),
                        e);
                Monitors.error(CLASS_NAME, "terminateWorkflow");
            }

            // update the failed reference task names
            List<TaskModel> failedTasks =
                    workflow.getTasks().stream()
                            .filter(
                                    t ->
                                            FAILED.equals(t.getStatus())
                                                    || FAILED_WITH_TERMINAL_ERROR.equals(
                                                            t.getStatus()))
                            .collect(Collectors.toList());

            workflow.getFailedReferenceTaskNames()
                    .addAll(
                            failedTasks.stream()
                                    .map(TaskModel::getReferenceTaskName)
                                    .collect(Collectors.toSet()));

            workflow.getFailedTaskNames()
                    .addAll(
                            failedTasks.stream()
                                    .map(TaskModel::getTaskDefName)
                                    .collect(Collectors.toSet()));

            String workflowId = workflow.getWorkflowId();
            workflow.setReasonForIncompletion(reason);
            // Cancel non-terminal tasks before updating workflow state and notifying the status
            // listener. The TERMINATED notification may trigger an archiving listener (e.g.
            // ArchivingWorkflowStatusListener) that immediately removes the workflow from the
            // primary data store. Archival requires tasks to be in a terminal state, so we must
            // cancel SCHEDULED/IN_PROGRESS tasks first.
            List<String> cancelErrors = cancelNonTerminalTasks(workflow);
            if (!cancelErrors.isEmpty()) {
                throw new NonTransientException(
                        String.format(
                                "Error canceling system tasks: %s",
                                String.join(",", cancelErrors)));
            }
            executionDAOFacade.updateWorkflow(workflow);
            notifyWorkflowStatusListener(workflow, WorkflowEventType.TERMINATED);
            Monitors.recordWorkflowTermination(
                    workflow.getWorkflowName(), workflow.getStatus(), workflow.getOwnerApp());
            LOGGER.info("Workflow {} is terminated because of {}", workflowId, reason);
            List<TaskModel> tasks = workflow.getTasks();
            try {
                // Remove from the task queue if they were there
                tasks.forEach(
                        task -> queueDAO.remove(QueueUtils.getQueueName(task), task.getTaskId()));
            } catch (Exception e) {
                LOGGER.warn(
                        "Error removing task(s) from queue during workflow termination : {}",
                        workflowId,
                        e);
            }

            if (workflow.hasParent()) {
                updateParentWorkflowTask(workflow);
                LOGGER.info(
                        "{} updated parent {} task {}",
                        workflow.toShortString(),
                        workflow.getParentWorkflowId(),
                        workflow.getParentWorkflowTaskId());
                expediteLazyWorkflowEvaluation(workflow.getParentWorkflowId());
            }

            if (!StringUtils.isBlank(failureWorkflow)) {
                Map<String, Object> input = new HashMap<>(workflow.getInput());
                input.put("workflowId", workflowId);
                input.put("reason", reason);
                input.put("failureStatus", workflow.getStatus().toString());
                if (workflow.getFailedTaskId() != null) {
                    input.put("failureTaskId", workflow.getFailedTaskId());
                }
                input.put("failedWorkflow", workflow);

                try {
                    String failureWFId = idGenerator.generate();
                    StartWorkflowInput startWorkflowInput = new StartWorkflowInput();
                    startWorkflowInput.setName(failureWorkflow);
                    startWorkflowInput.setVersion(failureWorkflowVersion);
                    startWorkflowInput.setWorkflowInput(input);
                    startWorkflowInput.setCorrelationId(workflow.getCorrelationId());
                    startWorkflowInput.setTaskToDomain(workflow.getTaskToDomain());
                    startWorkflowInput.setWorkflowId(failureWFId);
                    startWorkflowInput.setTriggeringWorkflowId(workflowId);

                    startWorkflow(startWorkflowInput);

                    workflow.addOutput("conductor.failure_workflow", failureWFId);
                } catch (Exception e) {
                    LOGGER.error("Failed to start error workflow", e);
                    workflow.getOutput()
                            .put(
                                    "conductor.failure_workflow",
                                    "Error workflow "
                                            + failureWorkflow
                                            + " failed to start.  reason: "
                                            + e.getMessage());
                    Monitors.recordWorkflowStartError(
                            failureWorkflow, WorkflowContext.get().getClientApp());
                }
                executionDAOFacade.updateWorkflow(workflow);
            }
            executionDAOFacade.removeFromPendingWorkflow(
                    workflow.getWorkflowName(), workflow.getWorkflowId());

            workflowMessageQueueDAO.ifPresent(dao -> dao.delete(workflow.getWorkflowId()));
            return workflow;
        } finally {
            executionLockService.releaseLock(workflow.getWorkflowId());
            executionLockService.deleteLock(workflow.getWorkflowId());
        }
    }

    /**
     * @param taskResult the task result to be updated.
     * @throws IllegalArgumentException if the {@link TaskResult} is null. @Returns Updated task
     * @throws NotFoundException if the Task is not found.
     */
    @Override
    public TaskModel updateTask(TaskResult taskResult) {
        if (taskResult == null) {
            throw new IllegalArgumentException("Task object is null");
        } else if (taskResult.isExtendLease()) {
            extendLease(taskResult);
            return null;
        }

        String workflowId = taskResult.getWorkflowInstanceId();
        WorkflowModel workflowInstance = executionDAOFacade.getWorkflowModel(workflowId, false);

        TaskModel task =
                Optional.ofNullable(executionDAOFacade.getTaskModel(taskResult.getTaskId()))
                        .orElseThrow(
                                () ->
                                        new NotFoundException(
                                                "No such task found by id: %s",
                                                taskResult.getTaskId()));

        LOGGER.debug("Task: {} belonging to Workflow {} being updated", task, workflowInstance);

        String taskQueueName = QueueUtils.getQueueName(task);

        if (task.getStatus().isTerminal()) {
            // Task was already updated....
            queueDAO.remove(taskQueueName, taskResult.getTaskId());
            LOGGER.info(
                    "Task: {} has already finished execution with status: {} within workflow: {}. Removed task from queue: {}",
                    task.getTaskId(),
                    task.getStatus(),
                    task.getWorkflowInstanceId(),
                    taskQueueName);
            Monitors.recordUpdateConflict(
                    task.getTaskType(), workflowInstance.getWorkflowName(), task.getStatus());
            return task;
        }

        if (workflowInstance.getStatus().isTerminal()) {
            // Workflow is in terminal state
            queueDAO.remove(taskQueueName, taskResult.getTaskId());
            LOGGER.info(
                    "Workflow: {} has already finished execution. Task update for: {} ignored and removed from Queue: {}.",
                    workflowInstance,
                    taskResult.getTaskId(),
                    taskQueueName);
            Monitors.recordUpdateConflict(
                    task.getTaskType(),
                    workflowInstance.getWorkflowName(),
                    workflowInstance.getStatus());
            return task;
        }

        // for system tasks, setting to SCHEDULED would mean restarting the task which
        // is
        // undesirable
        // for worker tasks, set status to SCHEDULED and push to the queue
        if (!systemTaskRegistry.isSystemTask(task.getTaskType())
                && taskResult.getStatus() == TaskResult.Status.IN_PROGRESS) {
            task.setStatus(SCHEDULED);
        } else {
            task.setStatus(TaskModel.Status.valueOf(taskResult.getStatus().name()));
        }
        task.setOutputMessage(taskResult.getOutputMessage());
        task.setReasonForIncompletion(taskResult.getReasonForIncompletion());
        task.setWorkerId(taskResult.getWorkerId());
        task.setCallbackAfterSeconds(taskResult.getCallbackAfterSeconds());
        task.setOutputData(taskResult.getOutputData());
        task.setSubWorkflowId(taskResult.getSubWorkflowId());

        if (StringUtils.isNotBlank(taskResult.getExternalOutputPayloadStoragePath())) {
            task.setExternalOutputPayloadStoragePath(
                    taskResult.getExternalOutputPayloadStoragePath());
        }

        if (task.getStatus().isTerminal()) {
            task.setEndTime(System.currentTimeMillis());
        }

        // Update message in Task queue based on Task status
        switch (task.getStatus()) {
            case COMPLETED:
            case CANCELED:
            case FAILED:
            case FAILED_WITH_TERMINAL_ERROR:
            case TIMED_OUT:
                try {
                    queueDAO.remove(taskQueueName, taskResult.getTaskId());
                    LOGGER.debug(
                            "Task: {} removed from taskQueue: {} since the task status is {}",
                            task,
                            taskQueueName,
                            task.getStatus().name());
                } catch (Exception e) {
                    // Ignore exceptions on queue remove as it wouldn't impact task and workflow
                    // execution, and will be cleaned up eventually
                    String errorMsg =
                            String.format(
                                    "Error removing the message in queue for task: %s for workflow: %s",
                                    task.getTaskId(), workflowId);
                    LOGGER.warn(errorMsg, e);
                    Monitors.recordTaskQueueOpError(
                            task.getTaskType(), workflowInstance.getWorkflowName());
                }
                break;
            case IN_PROGRESS:
            case SCHEDULED:
                try {
                    long callBack = taskResult.getCallbackAfterSeconds();
                    queueDAO.postpone(
                            taskQueueName, task.getTaskId(), task.getWorkflowPriority(), callBack);
                    LOGGER.debug(
                            "Task: {} postponed in taskQueue: {} since the task status is {} with callbackAfterSeconds: {}",
                            task,
                            taskQueueName,
                            task.getStatus().name(),
                            callBack);
                } catch (Exception e) {
                    // Throw exceptions on queue postpone, this would impact task execution
                    String errorMsg =
                            String.format(
                                    "Error postponing the message in queue for task: %s for workflow: %s",
                                    task.getTaskId(), workflowId);
                    LOGGER.error(errorMsg, e);
                    Monitors.recordTaskQueueOpError(
                            task.getTaskType(), workflowInstance.getWorkflowName());
                    throw new TransientException(errorMsg, e);
                }
                break;
            default:
                break;
        }

        // Throw a TransientException if below operations fail to avoid workflow
        // inconsistencies.
        try {
            executionDAOFacade.updateTask(task);
        } catch (Exception e) {
            String errorMsg =
                    String.format(
                            "Error updating task: %s for workflow: %s",
                            task.getTaskId(), workflowId);
            LOGGER.error(errorMsg, e);
            Monitors.recordTaskUpdateError(task.getTaskType(), workflowInstance.getWorkflowName());
            throw new TransientException(errorMsg, e);
        }

        try {
            notifyTaskStatusListener(task);
        } catch (Exception e) {
            String errorMsg =
                    String.format(
                            "Error while notifying TaskStatusListener: %s for workflow: %s",
                            task.getTaskId(), workflowId);
            LOGGER.error(errorMsg, e);
        }

        List<TaskExecLog> taskLogs = taskResult.getLogs();
        if (taskLogs != null) {
            taskLogs.forEach(taskExecLog -> taskExecLog.setTaskId(task.getTaskId()));
            executionDAOFacade.addTaskExecLog(taskLogs);
        }

        if (task.getStatus().isTerminal()) {
            long duration = getTaskDuration(0, task);
            long lastDuration = task.getEndTime() - task.getStartTime();
            Monitors.recordTaskExecutionTime(
                    task.getTaskDefName(), duration, true, task.getStatus());
            Monitors.recordTaskExecutionTime(
                    task.getTaskDefName(), lastDuration, false, task.getStatus());
        }

        if (properties.isHumanTaskPreventsDeciderQueue()
                && TaskType.TASK_TYPE_HUMAN.equals(task.getTaskType())
                && task.getStatus().isTerminal()) {
            // The sweeper removes workflows blocked on a HUMAN task from the decider queue.
            // Re-queue this workflow so it is evaluated immediately now that the HUMAN task has
            // reached a terminal status.
            queueDAO.push(DECIDER_QUEUE, workflowId, workflowInstance.getPriority(), 0);
            LOGGER.debug(
                    "Waking up workflow {} because HUMAN task {} reached terminal status {}",
                    workflowId,
                    task.getTaskId(),
                    task.getStatus());
        }

        if (!isLazyEvaluateWorkflow(workflowInstance.getWorkflowDefinition(), task)) {
            decide(workflowId);
        }
        return task;
    }

    private void notifyTaskStatusListener(TaskModel task) {
        switch (task.getStatus()) {
            case COMPLETED:
                taskStatusListener.onTaskCompletedIfEnabled(task);
                break;
            case CANCELED:
                taskStatusListener.onTaskCanceledIfEnabled(task);
                break;
            case FAILED:
                taskStatusListener.onTaskFailedIfEnabled(task);
                break;
            case FAILED_WITH_TERMINAL_ERROR:
                taskStatusListener.onTaskFailedWithTerminalErrorIfEnabled(task);
                break;
            case TIMED_OUT:
                taskStatusListener.onTaskTimedOutIfEnabled(task);
                break;
            case IN_PROGRESS:
                taskStatusListener.onTaskInProgressIfEnabled(task);
                break;
            case SCHEDULED:
                // no-op, already done in addTaskToQueue
            default:
                break;
        }
    }

    private void extendLease(TaskResult taskResult) {
        TaskModel task =
                Optional.ofNullable(executionDAOFacade.getTaskModel(taskResult.getTaskId()))
                        .orElseThrow(
                                () ->
                                        new NotFoundException(
                                                "No such task found by id: %s",
                                                taskResult.getTaskId()));

        LOGGER.debug(
                "Extend lease for Task: {} belonging to Workflow: {}",
                task,
                task.getWorkflowInstanceId());
        if (!task.getStatus().isTerminal()) {
            try {
                executionDAOFacade.extendLease(task);
            } catch (Exception e) {
                String errorMsg =
                        String.format(
                                "Error extend lease for Task: %s belonging to Workflow: %s",
                                task.getTaskId(), task.getWorkflowInstanceId());
                LOGGER.error(errorMsg, e);
                Monitors.recordTaskExtendLeaseError(task.getTaskType(), task.getWorkflowType());
                throw new TransientException(errorMsg, e);
            }
        }
    }

    /**
     * Determines if a workflow can be lazily evaluated, if it meets any of these criteria
     *
     * <ul>
     *   <li>The task is NOT a loop task within DO_WHILE
     *   <li>The task is one of the intermediate tasks in a branch within a FORK_JOIN
     *   <li>The task is forked from a FORK_JOIN_DYNAMIC
     * </ul>
     *
     * @param workflowDef The workflow definition of the workflow for which evaluation decision is
     *     to be made
     * @param task The task which is attempting to trigger the evaluation
     * @return true if workflow can be lazily evaluated, false otherwise
     */
    @VisibleForTesting
    boolean isLazyEvaluateWorkflow(WorkflowDef workflowDef, TaskModel task) {
        if (task.isLoopOverTask()) {
            return false;
        }

        String taskRefName = task.getReferenceTaskName();
        List<WorkflowTask> workflowTasks = workflowDef.collectTasks();

        List<WorkflowTask> forkTasks =
                workflowTasks.stream()
                        .filter(t -> t.getType().equals(TaskType.FORK_JOIN.name()))
                        .collect(Collectors.toList());

        List<WorkflowTask> joinTasks =
                workflowTasks.stream()
                        .filter(t -> t.getType().equals(TaskType.JOIN.name()))
                        .collect(Collectors.toList());

        if (forkTasks.stream().anyMatch(fork -> fork.has(taskRefName))) {
            return joinTasks.stream().anyMatch(join -> join.getJoinOn().contains(taskRefName))
                    && task.getStatus().isSuccessful();
        }

        // Tasks not in the workflow definition are dynamically forked (FORK_JOIN_DYNAMIC).
        // Always trigger decide for these tasks: they rely on the sweeper which has a
        // 30+ second initial delay, causing workflows to stall if decide is skipped.
        return false;
    }

    @Override
    public TaskModel getTask(String taskId) {
        return Optional.ofNullable(executionDAOFacade.getTaskModel(taskId))
                .map(
                        task -> {
                            if (task.getWorkflowTask() != null) {
                                return metadataMapperService.populateTaskWithDefinition(task);
                            }
                            return task;
                        })
                .orElse(null);
    }

    @Override
    public List<Workflow> getRunningWorkflows(String workflowName, int version) {
        return executionDAOFacade.getPendingWorkflowsByName(workflowName, version);
    }

    @Override
    public List<String> getWorkflows(String name, Integer version, Long startTime, Long endTime) {
        return executionDAOFacade.getWorkflowsByName(name, startTime, endTime).stream()
                .filter(workflow -> workflow.getWorkflowVersion() == version)
                .map(Workflow::getWorkflowId)
                .collect(Collectors.toList());
    }

    @Override
    public List<String> getRunningWorkflowIds(String workflowName, int version) {
        return executionDAOFacade.getRunningWorkflowIds(workflowName, version);
    }

    /** Records a metric for the "decide" process. */
    @Override
    public WorkflowModel decide(String workflowId) {
        StopWatch watch = new StopWatch();
        watch.start();
        boolean lockAcquired = executionLockService.acquireLock(workflowId);
        if (!lockAcquired) {
            return null;
        }
        try {

            WorkflowModel workflow = executionDAOFacade.getWorkflowModel(workflowId, true);
            if (workflow == null) {
                // This can happen if the workflowId is incorrect
                return null;
            }
            return decide(workflow);

        } finally {
            if (lockAcquired) {
                executionLockService.releaseLock(workflowId);
            }
            watch.stop();
            Monitors.recordWorkflowDecisionTime(watch.getTime());
        }
    }

    @Override
    public WorkflowModel decideWithLock(WorkflowModel workflow) {
        if (!executionLockService.acquireLock(workflow.getWorkflowId())) {
            LOGGER.debug(
                    "decideWithLock couldn't acquire lock for workflow {}",
                    workflow.getWorkflowId());
            return null;
        }
        try {
            return decide(workflow);
        } finally {
            executionLockService.releaseLock(workflow.getWorkflowId());
        }
    }

    /**
     * @param workflow the workflow to evaluate the state for
     * @return true if the workflow has completed (success or failed), false otherwise. Note: This
     *     method does not acquire the lock on the workflow and should ony be called / overridden if
     *     No locking is required or lock is acquired externally
     */
    private WorkflowModel decide(WorkflowModel workflow) {
        if (workflow.getStatus().isTerminal()) {
            if (!workflow.getStatus().isSuccessful()) {
                cancelNonTerminalTasks(workflow);
            }
            return workflow;
        }

        // we find any sub workflow tasks that have changed
        // and change the workflow/task state accordingly
        adjustStateIfSubWorkflowChanged(workflow);

        // Guard against holding the lock past its lease time. If synchronous system tasks
        // (e.g. INLINE inside a DO_WHILE) keep changing state, we loop instead of recursing to
        // avoid a StackOverflowError. When the lease is about to expire we break out, persist the
        // current state, and re-queue the workflow so the sweeper picks it up cleanly.
        final long maxRuntime = properties.getLockLeaseTime().toMillis() - 100;
        StopWatch decideWatch = new StopWatch();
        decideWatch.start();

        try {
            boolean continueLoop = true;
            while (continueLoop) {
                continueLoop = false;

                DeciderService.DeciderOutcome outcome = deciderService.decide(workflow);
                if (outcome.isComplete) {
                    endExecution(workflow, outcome.terminateTask);
                    return workflow;
                }

                List<TaskModel> tasksToBeScheduled = outcome.tasksToBeScheduled;
                setTaskDomains(tasksToBeScheduled, workflow);
                List<TaskModel> tasksToBeUpdated = outcome.tasksToBeUpdated;

                tasksToBeScheduled = dedupAndAddTasks(workflow, tasksToBeScheduled);

                boolean stateChanged = scheduleTask(workflow, tasksToBeScheduled);

                for (TaskModel task : outcome.tasksToBeScheduled) {
                    executionDAOFacade.populateTaskData(task);
                    if (systemTaskRegistry.isSystemTask(task.getTaskType())
                            && NON_TERMINAL_TASK.test(task)) {
                        WorkflowSystemTask workflowSystemTask =
                                systemTaskRegistry.get(task.getTaskType());
                        if (!workflowSystemTask.isAsync()
                                && workflowSystemTask.execute(workflow, task, this)) {
                            tasksToBeUpdated.add(task);
                            stateChanged = true;
                        }
                    }
                }

                if (!outcome.tasksToBeUpdated.isEmpty() || !tasksToBeScheduled.isEmpty()) {
                    executionDAOFacade.updateTasks(tasksToBeUpdated);
                }

                if (stateChanged) {
                    if (decideWatch.getTime() < maxRuntime) {
                        continueLoop = true;
                        continue;
                    }
                    // Lock lease is about to expire. Persist current state and re-queue so
                    // the next decide() cycle continues without holding a stale lock.
                    LOGGER.info(
                            "Workflow {} decide loop approaching lock lease time after {} ms, "
                                    + "re-queuing for continued processing",
                            workflow.getWorkflowId(),
                            decideWatch.getTime());
                    executionDAOFacade.updateWorkflow(workflow);
                    queueDAO.push(DECIDER_QUEUE, workflow.getWorkflowId(), 0);
                    return workflow;
                }

                if (!outcome.tasksToBeUpdated.isEmpty() || !tasksToBeScheduled.isEmpty()) {
                    executionDAOFacade.updateWorkflow(workflow);
                }

                Duration timeout = properties.getWorkflowOffsetTimeout();
                if (!workflow.getStatus().isTerminal()) {
                    if (properties.isHumanTaskPreventsDeciderQueue()
                            && hasInProgressHumanTask(workflow)) {
                        // A workflow blocked on a HUMAN task can wait indefinitely for external
                        // input. Keeping it in the decider queue only churns the sweeper, so
                        // remove it; updateTask() re-queues the workflow when the HUMAN task
                        // reaches a terminal status.
                        LOGGER.debug(
                                "Removing workflow {} from decider queue; blocked on HUMAN task",
                                workflow.getWorkflowId());
                        queueDAO.remove(DECIDER_QUEUE, workflow.getWorkflowId());
                    } else {
                        Duration updatedOffset =
                                computePostpone(
                                        workflow,
                                        timeout,
                                        properties.getMaxPostponeDurationSeconds());
                        if (updatedOffset.getSeconds() != timeout.getSeconds()) {
                            // we have a new value, setUnack uses time in millis
                            LOGGER.debug(
                                    "Pushing the workflow {} into decider queue by {} millis",
                                    workflow.getWorkflowId(),
                                    updatedOffset.getSeconds() * 1000);
                            queueDAO.setUnackTimeout(
                                    DECIDER_QUEUE,
                                    workflow.getWorkflowId(),
                                    updatedOffset.getSeconds() * 1000);
                        }
                    }
                }
            }

            return workflow;

        } catch (TerminateWorkflowException twe) {
            LOGGER.info("Execution terminated of workflow: {}", workflow, twe);
            terminate(workflow, twe);
            return workflow;
        } catch (RuntimeException e) {
            LOGGER.error("Error deciding workflow: {}", workflow.getWorkflowId(), e);
            throw e;
        }
    }

    private void adjustStateIfSubWorkflowChanged(WorkflowModel workflow) {
        Optional<TaskModel> changedSubWorkflowTask = findChangedSubWorkflowTask(workflow);
        if (changedSubWorkflowTask.isPresent()) {
            // reset the flag
            TaskModel subWorkflowTask = changedSubWorkflowTask.get();
            subWorkflowTask.setSubworkflowChanged(false);
            executionDAOFacade.updateTask(subWorkflowTask);

            LOGGER.info(
                    "{} reset subworkflowChanged flag for {}",
                    workflow.toShortString(),
                    subWorkflowTask.getTaskId());

            // find all terminal and unsuccessful JOIN tasks and set them to IN_PROGRESS
            // if we are here, then the SUB_WORKFLOW task could be part of a FORK_JOIN or
            // FORK_JOIN_DYNAMIC and the JOIN task(s) needs to be evaluated again, set them to
            // IN_PROGRESS
            resetUnsuccessfulJoinTasks(workflow);
        }
    }

    /**
     * Re-queue every IN_PROGRESS JOIN in {@code parentWorkflowId} for immediate re-evaluation.
     * Called when a fork branch's sub-workflow reaches a terminal state: the sibling JOIN may now
     * be satisfiable, but as an async task it only re-polls on exponential backoff (capped at
     * workflowOffsetTimeout, default 30s). Without this nudge the parent can hang for up to that
     * cap after the last branch finishes before the JOIN notices. Mirrors {@link
     * #expediteLazyWorkflowEvaluation}: postpone the existing queue message to 0 if present, else
     * push a fresh one — idempotent by task id, so no duplicate queue entries.
     */
    private void expediteInProgressJoinTasks(String parentWorkflowId) {
        WorkflowModel parentWorkflow = executionDAOFacade.getWorkflowModel(parentWorkflowId, true);
        if (parentWorkflow == null) {
            return;
        }
        for (TaskModel joinTask : parentWorkflow.getTasks()) {
            if (!TaskType.TASK_TYPE_JOIN.equals(joinTask.getTaskType())
                    || joinTask.getStatus() != TaskModel.Status.IN_PROGRESS) {
                continue;
            }
            String queueName = QueueUtils.getQueueName(joinTask);
            if (queueDAO.containsMessage(queueName, joinTask.getTaskId())) {
                queueDAO.postpone(
                        queueName, joinTask.getTaskId(), joinTask.getWorkflowPriority(), 0);
            } else {
                queueDAO.push(queueName, joinTask.getTaskId(), joinTask.getWorkflowPriority(), 0);
            }
        }
    }

    private Optional<TaskModel> findChangedSubWorkflowTask(WorkflowModel workflow) {
        WorkflowDef workflowDef =
                Optional.ofNullable(workflow.getWorkflowDefinition())
                        .orElseGet(
                                () ->
                                        metadataDAO
                                                .getWorkflowDef(
                                                        workflow.getWorkflowName(),
                                                        workflow.getWorkflowVersion())
                                                .orElseThrow(
                                                        () ->
                                                                new TransientException(
                                                                        "Workflow Definition is not found")));
        if (workflowDef.containsType(TaskType.TASK_TYPE_SUB_WORKFLOW)
                || workflow.getWorkflowDefinition()
                        .containsType(TaskType.TASK_TYPE_FORK_JOIN_DYNAMIC)) {
            return workflow.getTasks().stream()
                    .filter(
                            t ->
                                    t.getTaskType().equals(TaskType.TASK_TYPE_SUB_WORKFLOW)
                                            && t.isSubworkflowChanged()
                                            && !t.isRetried())
                    .findFirst();
        }
        return Optional.empty();
    }

    @VisibleForTesting
    List<String> cancelNonTerminalTasks(WorkflowModel workflow) {
        return cancelNonTerminalTasks(workflow, true);
    }

    List<String> cancelNonTerminalTasks(WorkflowModel workflow, boolean raiseFinalized) {
        List<String> erroredTasks = new ArrayList<>();
        // Update non-terminal tasks' status to CANCELED
        for (TaskModel task : workflow.getTasks()) {
            if (!task.getStatus().isTerminal()) {
                // Cancel the ones which are not completed yet....
                task.setStatus(CANCELED);
                try {
                    notifyTaskStatusListener(task);
                } catch (Exception e) {
                    String errorMsg =
                            String.format(
                                    "Error while notifying TaskStatusListener: %s for workflow: %s",
                                    task.getTaskId(), task.getWorkflowInstanceId());
                    LOGGER.error(errorMsg, e);
                }
                if (systemTaskRegistry.isSystemTask(task.getTaskType())) {
                    WorkflowSystemTask workflowSystemTask =
                            systemTaskRegistry.get(task.getTaskType());
                    try {
                        workflowSystemTask.cancel(workflow, task, this);
                    } catch (Exception e) {
                        erroredTasks.add(task.getReferenceTaskName());
                        LOGGER.error(
                                "Error canceling system task:{}/{} in workflow: {}",
                                workflowSystemTask.getTaskType(),
                                task.getTaskId(),
                                workflow.getWorkflowId(),
                                e);
                    }
                }
                executionDAOFacade.updateTask(task);
            }
        }
        if (erroredTasks.isEmpty()) {
            try {
                if (raiseFinalized) {
                    notifyWorkflowStatusListener(workflow, WorkflowEventType.FINALIZED);
                }
                queueDAO.remove(DECIDER_QUEUE, workflow.getWorkflowId());
            } catch (Exception e) {
                LOGGER.error(
                        "Error removing workflow: {} from decider queue",
                        workflow.getWorkflowId(),
                        e);
            }
        }
        return erroredTasks;
    }

    @VisibleForTesting
    List<TaskModel> dedupAndAddTasks(WorkflowModel workflow, List<TaskModel> tasks) {
        Set<String> tasksInWorkflow =
                workflow.getTasks().stream()
                        .map(task -> task.getReferenceTaskName() + "_" + task.getRetryCount())
                        .collect(Collectors.toSet());

        List<TaskModel> dedupedTasks =
                tasks.stream()
                        .filter(
                                task ->
                                        !tasksInWorkflow.contains(
                                                task.getReferenceTaskName()
                                                        + "_"
                                                        + task.getRetryCount()))
                        .collect(Collectors.toList());

        workflow.getTasks().addAll(dedupedTasks);
        return dedupedTasks;
    }

    /**
     * @throws ConflictException if the workflow is in terminal state.
     */
    @Override
    public void pauseWorkflow(String workflowId) {
        try {
            executionLockService.acquireLock(workflowId, 60000);
            WorkflowModel.Status status = WorkflowModel.Status.PAUSED;
            WorkflowModel workflow = executionDAOFacade.getWorkflowModel(workflowId, false);
            if (workflow.getStatus().isTerminal()) {
                throw new ConflictException(
                        "Workflow %s has ended, status cannot be updated.",
                        workflow.toShortString());
            }
            if (workflow.getStatus().equals(status)) {
                return; // Already paused!
            }
            workflow.setStatus(status);
            executionDAOFacade.updateWorkflow(workflow);

            // Notify on workflow paused.
            notifyWorkflowStatusListener(workflow, WorkflowEventType.PAUSED);
        } finally {
            executionLockService.releaseLock(workflowId);
        }

        // remove from the sweep queue
        // any exceptions can be ignored, as this is not critical to the pause operation
        try {
            queueDAO.remove(DECIDER_QUEUE, workflowId);
        } catch (Exception e) {
            LOGGER.info(
                    "[pauseWorkflow] Error removing workflow: {} from decider queue",
                    workflowId,
                    e);
        }
    }

    /**
     * @param workflowId the workflow to be resumed
     * @throws IllegalStateException if the workflow is not in PAUSED state
     */
    @Override
    public void resumeWorkflow(String workflowId) {
        WorkflowModel workflow = executionDAOFacade.getWorkflowModel(workflowId, false);
        if (!workflow.getStatus().equals(WorkflowModel.Status.PAUSED)) {
            throw new IllegalStateException(
                    "The workflow "
                            + workflowId
                            + " is not PAUSED so cannot resume. "
                            + "Current status is "
                            + workflow.getStatus().name());
        }
        workflow.setStatus(WorkflowModel.Status.RUNNING);
        workflow.setLastRetriedTime(System.currentTimeMillis());
        // Add to decider queue
        queueDAO.push(
                DECIDER_QUEUE,
                workflow.getWorkflowId(),
                workflow.getPriority(),
                properties.getWorkflowOffsetTimeout().getSeconds());
        executionDAOFacade.updateWorkflow(workflow);
        // Notify on workflow resumed.
        notifyWorkflowStatusListener(workflow, WorkflowEventType.RESUMED);
        decide(workflowId);
    }

    /**
     * @param workflowId the id of the workflow
     * @param taskReferenceName the referenceName of the task to be skipped
     * @param skipTaskRequest the {@link SkipTaskRequest} object
     * @throws IllegalStateException
     */
    @Override
    public void skipTaskFromWorkflow(
            String workflowId, String taskReferenceName, SkipTaskRequest skipTaskRequest) {

        WorkflowModel workflow = executionDAOFacade.getWorkflowModel(workflowId, true);

        // If the workflow is not running then cannot skip any task
        if (!workflow.getStatus().equals(WorkflowModel.Status.RUNNING)) {
            String errorMsg =
                    String.format(
                            "The workflow %s is not running so the task referenced by %s cannot be skipped",
                            workflowId, taskReferenceName);
            throw new IllegalStateException(errorMsg);
        }

        // Check if the reference name is as per the workflowdef
        WorkflowTask workflowTask =
                workflow.getWorkflowDefinition().getTaskByRefName(taskReferenceName);
        if (workflowTask == null) {
            String errorMsg =
                    String.format(
                            "The task referenced by %s does not exist in the WorkflowDefinition %s",
                            taskReferenceName, workflow.getWorkflowName());
            throw new IllegalStateException(errorMsg);
        }

        // If the task is already started the again it cannot be skipped
        workflow.getTasks()
                .forEach(
                        task -> {
                            if (task.getReferenceTaskName().equals(taskReferenceName)) {
                                String errorMsg =
                                        String.format(
                                                "The task referenced %s has already been processed, cannot be skipped",
                                                taskReferenceName);
                                throw new IllegalStateException(errorMsg);
                            }
                        });

        // Now create a "SKIPPED" task for this workflow
        TaskModel taskToBeSkipped = new TaskModel();
        taskToBeSkipped.setTaskId(idGenerator.generate());
        taskToBeSkipped.setReferenceTaskName(taskReferenceName);
        taskToBeSkipped.setWorkflowInstanceId(workflowId);
        taskToBeSkipped.setWorkflowPriority(workflow.getPriority());
        taskToBeSkipped.setStatus(SKIPPED);
        taskToBeSkipped.setEndTime(System.currentTimeMillis());
        taskToBeSkipped.setTaskType(workflowTask.getName());
        taskToBeSkipped.setCorrelationId(workflow.getCorrelationId());
        if (skipTaskRequest != null) {
            taskToBeSkipped.setInputData(skipTaskRequest.getTaskInput());
            taskToBeSkipped.setOutputData(skipTaskRequest.getTaskOutput());
            taskToBeSkipped.setInputMessage(skipTaskRequest.getTaskInputMessage());
            taskToBeSkipped.setOutputMessage(skipTaskRequest.getTaskOutputMessage());
        }
        executionDAOFacade.createTasks(Collections.singletonList(taskToBeSkipped));
        decide(workflow.getWorkflowId());
    }

    @Override
    public WorkflowModel getWorkflow(String workflowId, boolean includeTasks) {
        return executionDAOFacade.getWorkflowModel(workflowId, includeTasks);
    }

    private void addTaskToQueue(TaskModel task) {
        // put in queue
        String taskQueueName = QueueUtils.getQueueName(task);
        if (task.getCallbackAfterMs() > 0) {
            // Use ms-precision Duration overload to preserve sub-second jitter
            queueDAO.push(
                    taskQueueName,
                    task.getTaskId(),
                    task.getWorkflowPriority(),
                    Duration.ofMillis(task.getCallbackAfterMs()));
        } else if (task.getCallbackAfterSeconds() > 0) {
            queueDAO.push(
                    taskQueueName,
                    task.getTaskId(),
                    task.getWorkflowPriority(),
                    task.getCallbackAfterSeconds());
        } else {
            queueDAO.push(taskQueueName, task.getTaskId(), task.getWorkflowPriority(), 0);
        }
        LOGGER.debug(
                "Added task {} with priority {} to queue {} with call back seconds {}",
                task,
                task.getWorkflowPriority(),
                taskQueueName,
                task.getCallbackAfterSeconds());
    }

    @VisibleForTesting
    void setTaskDomains(List<TaskModel> tasks, WorkflowModel workflow) {
        Map<String, String> taskToDomain = workflow.getTaskToDomain();
        if (taskToDomain != null) {
            // Step 1: Apply * mapping to all tasks, if present.
            String domainstr = taskToDomain.get("*");
            if (StringUtils.isNotBlank(domainstr)) {
                String[] domains = domainstr.split(",");
                tasks.forEach(
                        task -> {
                            // Filter out SystemTask
                            if (!systemTaskRegistry.isSystemTask(task.getTaskType())) {
                                // Check which domain worker is polling
                                // Set the task domain
                                task.setDomain(getActiveDomain(task.getTaskType(), domains));
                            }
                        });
            }
            // Step 2: Override additional mappings.
            tasks.forEach(
                    task -> {
                        if (!systemTaskRegistry.isSystemTask(task.getTaskType())) {
                            String taskDomainstr = taskToDomain.get(task.getTaskType());
                            if (taskDomainstr != null) {
                                task.setDomain(
                                        getActiveDomain(
                                                task.getTaskType(), taskDomainstr.split(",")));
                            }
                        }
                    });
        }
    }

    /**
     * Gets the active domain from the list of domains where the task is to be queued. The domain
     * list must be ordered. In sequence, check if any worker has polled for last
     * `activeWorkerLastPollMs`, if so that is the Active domain. When no active domains are found:
     * <li>If NO_DOMAIN token is provided, return null.
     * <li>Else, return last domain from list.
     *
     * @param taskType the taskType of the task for which active domain is to be found
     * @param domains the array of domains for the task. (Must contain atleast one element).
     * @return the active domain where the task will be queued
     */
    @VisibleForTesting
    String getActiveDomain(String taskType, String[] domains) {
        if (domains == null || domains.length == 0) {
            return null;
        }

        return Arrays.stream(domains)
                .filter(domain -> !domain.equalsIgnoreCase("NO_DOMAIN"))
                .map(domain -> executionDAOFacade.getTaskPollDataByDomain(taskType, domain.trim()))
                .filter(Objects::nonNull)
                .filter(validateLastPolledTime)
                .findFirst()
                .map(PollData::getDomain)
                .orElse(
                        domains[domains.length - 1].trim().equalsIgnoreCase("NO_DOMAIN")
                                ? null
                                : domains[domains.length - 1].trim());
    }

    private long getTaskDuration(long s, TaskModel task) {
        long duration = task.getEndTime() - task.getStartTime();
        s += duration;
        if (task.getRetriedTaskId() == null) {
            return s;
        }
        return s + getTaskDuration(s, executionDAOFacade.getTaskModel(task.getRetriedTaskId()));
    }

    @VisibleForTesting
    boolean scheduleTask(WorkflowModel workflow, List<TaskModel> tasks) {
        List<TaskModel> tasksToBeQueued;
        boolean startedSystemTasks = false;

        try {
            if (tasks == null || tasks.isEmpty()) {
                return false;
            }

            // Get the highest seq number
            int count = workflow.getTasks().stream().mapToInt(TaskModel::getSeq).max().orElse(0);

            for (TaskModel task : tasks) {
                if (task.getSeq() == 0) { // Set only if the seq was not set
                    task.setSeq(++count);
                }
                // Stamp the very first time this task enters the queue. Retried tasks carry
                // this value forward (via TaskModel.copy()), so it is never overwritten here.
                if (task.getFirstScheduledTime() == 0) {
                    task.setFirstScheduledTime(System.currentTimeMillis());
                }
            }

            // metric to track the distribution of number of tasks within a workflow
            Monitors.recordNumTasksInWorkflow(
                    workflow.getTasks().size() + tasks.size(),
                    workflow.getWorkflowName(),
                    String.valueOf(workflow.getWorkflowVersion()));

            // Save the tasks in the DAO
            executionDAOFacade.createTasks(tasks);

            List<TaskModel> systemTasks =
                    tasks.stream()
                            .filter(task -> systemTaskRegistry.isSystemTask(task.getTaskType()))
                            .collect(Collectors.toList());

            tasksToBeQueued =
                    tasks.stream()
                            .filter(task -> !systemTaskRegistry.isSystemTask(task.getTaskType()))
                            .collect(Collectors.toList());

            // Traverse through all the system tasks, start the sync tasks, in case of async
            // queue
            // the tasks
            for (TaskModel task : systemTasks) {
                WorkflowSystemTask workflowSystemTask = systemTaskRegistry.get(task.getTaskType());
                if (workflowSystemTask == null) {
                    throw new NotFoundException(
                            "No system task found by name %s", task.getTaskType());
                }
                if (task.getStatus() != null
                        && !task.getStatus().isTerminal()
                        && task.getStartTime() == 0) {
                    task.setStartTime(System.currentTimeMillis());
                }
                if (!workflowSystemTask.isAsync()) {
                    try {
                        // start execution of synchronous system tasks
                        workflowSystemTask.start(workflow, task, this);
                    } catch (Exception e) {
                        String errorMsg =
                                String.format(
                                        "Unable to start system task: %s, {id: %s, name: %s}",
                                        task.getTaskType(),
                                        task.getTaskId(),
                                        task.getTaskDefName());
                        throw new NonTransientException(errorMsg, e);
                    }
                    startedSystemTasks = true;
                    executionDAOFacade.updateTask(task);
                } else {
                    tasksToBeQueued.add(task);
                }
            }

        } catch (Exception e) {
            List<String> taskIds =
                    tasks.stream().map(TaskModel::getTaskId).collect(Collectors.toList());
            String errorMsg =
                    String.format(
                            "Error scheduling tasks: %s, for workflow: %s",
                            taskIds, workflow.getWorkflowId());
            LOGGER.error(errorMsg, e);
            Monitors.error(CLASS_NAME, "scheduleTask");
            throw new TerminateWorkflowException(errorMsg);
        }

        // On addTaskToQueue failures, ignore the exceptions and let
        // WorkflowRepairService take care
        // of republishing the messages to the queue.
        try {
            addTaskToQueue(tasksToBeQueued);
        } catch (Exception e) {
            List<String> taskIds =
                    tasksToBeQueued.stream().map(TaskModel::getTaskId).collect(Collectors.toList());
            String errorMsg =
                    String.format(
                            "Error pushing tasks to the queue: %s, for workflow: %s",
                            taskIds, workflow.getWorkflowId());
            LOGGER.warn(errorMsg, e);
            Monitors.error(CLASS_NAME, "scheduleTask");
        }
        return startedSystemTasks;
    }

    private void addTaskToQueue(final List<TaskModel> tasks) {
        for (TaskModel task : tasks) {
            addTaskToQueue(task);
            // notify TaskStatusListener
            try {
                taskStatusListener.onTaskScheduledIfEnabled(task);
            } catch (Exception e) {
                String errorMsg =
                        String.format(
                                "Error while notifying TaskStatusListener: %s for workflow: %s",
                                task.getTaskId(), task.getWorkflowInstanceId());
                LOGGER.error(errorMsg, e);
            }
        }
    }

    private WorkflowModel terminate(
            final WorkflowModel workflow, TerminateWorkflowException terminateWorkflowException) {
        if (!workflow.getStatus().isTerminal()) {
            workflow.setStatus(terminateWorkflowException.getWorkflowStatus());
        }

        if (terminateWorkflowException.getTask() != null && workflow.getFailedTaskId() == null) {
            workflow.setFailedTaskId(terminateWorkflowException.getTask().getTaskId());
        }

        String failureWorkflow = workflow.getWorkflowDefinition().getFailureWorkflow();
        Integer failureWorkflowVersion =
                workflow.getWorkflowDefinition().getFailureWorkflowVersion();

        if (failureWorkflow != null) {
            if (failureWorkflow.startsWith("$")) {
                String[] paramPathComponents = failureWorkflow.split("\\.");
                String name = paramPathComponents[2]; // name of the input parameter
                failureWorkflow = (String) workflow.getInput().get(name);
            }
        }
        if (terminateWorkflowException.getTask() != null) {
            executionDAOFacade.updateTask(terminateWorkflowException.getTask());
        }
        return terminateWorkflow(
                workflow,
                terminateWorkflowException.getMessage(),
                failureWorkflow,
                failureWorkflowVersion);
    }

    private boolean rerunWF(
            String workflowId,
            String taskId,
            Map<String, Object> taskInput,
            Map<String, Object> workflowInput,
            String correlationId) {

        // Get the workflow
        WorkflowModel workflow = executionDAOFacade.getWorkflowModel(workflowId, true);
        if (!workflow.getStatus().isTerminal()) {
            String errorMsg =
                    String.format(
                            "Workflow: %s is not in terminal state, unable to rerun.", workflow);
            LOGGER.error(errorMsg);
            throw new ConflictException(errorMsg);
        }
        // If the task Id is null it implies that the entire workflow has to be rerun
        if (taskId == null) {
            // remove all tasks
            workflow.getTasks().forEach(task -> executionDAOFacade.removeTask(task.getTaskId()));
            workflow.setTasks(new ArrayList<>());
            // Set workflow as RUNNING
            workflow.setStatus(WorkflowModel.Status.RUNNING);
            // Reset failure reason from previous run to default
            workflow.setReasonForIncompletion(null);
            workflow.setFailedTaskId(null);
            workflow.setFailedReferenceTaskNames(new HashSet<>());
            workflow.setFailedTaskNames(new HashSet<>());

            if (correlationId != null) {
                workflow.setCorrelationId(correlationId);
            }
            if (workflowInput != null) {
                workflow.setInput(workflowInput);
            }

            queueDAO.push(
                    DECIDER_QUEUE,
                    workflow.getWorkflowId(),
                    workflow.getPriority(),
                    properties.getWorkflowOffsetTimeout().getSeconds());
            executionDAOFacade.updateWorkflow(workflow);
            updateAndPushParents(workflow, "reran");
            notifyWorkflowStatusListener(workflow, WorkflowEventType.RERAN);
            decide(workflowId);
            updateAndPushParents(workflow, "reran");
            return true;
        }

        // Now iterate through the tasks and find the "specific" task
        TaskModel rerunFromTask = null;
        for (TaskModel task : workflow.getTasks()) {
            if (task.getTaskId().equals(taskId)) {
                rerunFromTask = task;
                break;
            }
        }

        // If not found look into sub workflows
        if (rerunFromTask == null) {
            for (TaskModel task : workflow.getTasks()) {
                if (task.getTaskType().equalsIgnoreCase(TaskType.TASK_TYPE_SUB_WORKFLOW)) {
                    String subWorkflowId = task.getSubWorkflowId();
                    if (rerunWF(subWorkflowId, taskId, taskInput, null, null)) {
                        rerunFromTask = task;
                        break;
                    }
                }
            }
        }

        if (rerunFromTask != null) {
            // set workflow as RUNNING
            workflow.setStatus(WorkflowModel.Status.RUNNING);
            // Reset failure reason from previous run to default
            workflow.setReasonForIncompletion(null);
            workflow.setFailedTaskId(null);
            workflow.setFailedReferenceTaskNames(new HashSet<>());
            workflow.setFailedTaskNames(new HashSet<>());

            if (correlationId != null) {
                workflow.setCorrelationId(correlationId);
            }
            if (workflowInput != null) {
                workflow.setInput(workflowInput);
            }
            executionDAOFacade.updateWorkflow(workflow);
            // For direct non-SUB_WORKFLOW reruns, persist the target task as SCHEDULED before
            // updateAndPushParents fires expediteLazyWorkflowEvaluation. Without this early write,
            // an async decider that runs between the workflow-RUNNING write above and the
            // rerunFromTask-SCHEDULED write below can see all tasks as terminal and re-terminate
            // the workflow before PATH 3 gets a chance to reset rerunFromTask.
            if (rerunFromTask.getTaskId().equals(taskId)
                    && !TaskType.TASK_TYPE_SUB_WORKFLOW.equalsIgnoreCase(
                            rerunFromTask.getTaskType())) {
                rerunFromTask.setStatus(SCHEDULED);
                executionDAOFacade.updateTask(rerunFromTask);
            }
            updateAndPushParents(workflow, "reran");
            notifyWorkflowStatusListener(workflow, WorkflowEventType.RETRIED);

            // Recursive rerun targeting a SUB_WORKFLOW task: the child's finalizeRerun →
            // updateAndPushParents already wrote the correct states for this workflow's downstream
            // tasks (JOIN → IN_PROGRESS, sibling tasks → SCHEDULED) into the DB.
            // Writing the stale in-memory task list below would overwrite those correct DB values,
            // reverting JOIN to CANCELED. An async decider triggered by
            // expediteLazyWorkflowEvaluation would then see the CANCELED JOIN and terminate this
            // workflow. Skip the stale write, reset only rerunFromTask, then decide.
            if (!rerunFromTask.getTaskId().equals(taskId)
                    && rerunFromTask
                            .getTaskType()
                            .equalsIgnoreCase(TaskType.TASK_TYPE_SUB_WORKFLOW)) {
                rerunFromTask.setScheduledTime(System.currentTimeMillis());
                rerunFromTask.setStartTime(System.currentTimeMillis());
                rerunFromTask.setUpdateTime(0);
                rerunFromTask.setEndTime(0);
                rerunFromTask.setRetried(false);
                rerunFromTask.setExecuted(false);
                rerunFromTask.setPollCount(0);
                rerunFromTask.setStatus(IN_PROGRESS);
                rerunFromTask.setReasonForIncompletion(null);
                executionDAOFacade.updateTask(rerunFromTask);
                // Push AFTER task reset so async decider sees IN_PROGRESS, not stale FAILED state
                queueDAO.push(
                        DECIDER_QUEUE,
                        workflow.getWorkflowId(),
                        workflow.getPriority(),
                        properties.getWorkflowOffsetTimeout().getSeconds());
                decide(workflow.getWorkflowId());
                return true;
            }

            // update tasks in datastore to update workflow-tasks relationship for archived
            // workflows; exclude rerunFromTask, which is updated individually below — writing its
            // stale FAILED state here would race with the sweeper and can re-terminate the parent
            final String rerunTaskId = rerunFromTask.getTaskId();
            executionDAOFacade.updateTasks(
                    workflow.getTasks().stream()
                            .filter(t -> !t.getTaskId().equals(rerunTaskId))
                            .collect(Collectors.toList()));
            // Direct rerun targeting a SUB_WORKFLOW task: seq-based removal would strip parallel
            // fork branches — instead reset the task in-place, start a fresh child
            // synchronously so the task is IN_PROGRESS immediately, then let finalizeRerun
            // reset all terminal-unsuccessful siblings before the decider runs.
            if (rerunFromTask.getTaskId().equals(taskId)
                    && rerunFromTask
                            .getTaskType()
                            .equalsIgnoreCase(TaskType.TASK_TYPE_SUB_WORKFLOW)) {
                rerunFromTask.setScheduledTime(System.currentTimeMillis());
                rerunFromTask.setStartTime(0);
                rerunFromTask.setUpdateTime(0);
                rerunFromTask.setEndTime(0);
                rerunFromTask.clearOutput();
                rerunFromTask.setRetried(false);
                rerunFromTask.setExecuted(false);
                rerunFromTask.setPollCount(0);
                rerunFromTask.setSubWorkflowId(null);
                rerunFromTask.setStatus(SCHEDULED);
                rerunFromTask.setReasonForIncompletion(null);
                // Start the child workflow synchronously so the task is IN_PROGRESS before we
                // return — tests that read state immediately after rerun() need this.
                systemTaskRegistry
                        .get(TaskType.TASK_TYPE_SUB_WORKFLOW)
                        .start(workflow, rerunFromTask, this);
                if (rerunFromTask.getStatus() == SCHEDULED) {
                    // start() hit a transient error — fall back to async queue processing.
                    addTaskToQueue(rerunFromTask);
                }
                executionDAOFacade.updateTask(rerunFromTask);
                finalizeRerun(workflow, rerunFromTask);
                return true;
            }
            // Remove all tasks after the "rerunFromTask"
            List<TaskModel> filteredTasks = new ArrayList<>();
            for (TaskModel task : workflow.getTasks()) {
                if (task.getSeq() > rerunFromTask.getSeq()) {
                    executionDAOFacade.removeTask(task.getTaskId());
                } else {
                    filteredTasks.add(task);
                }
            }
            workflow.setTasks(filteredTasks);
            // reset fields before restarting the task
            rerunFromTask.setScheduledTime(System.currentTimeMillis());
            rerunFromTask.setStartTime(0);
            rerunFromTask.setUpdateTime(0);
            rerunFromTask.setEndTime(0);
            rerunFromTask.clearOutput();
            rerunFromTask.setRetried(false);
            rerunFromTask.setExecuted(false);
            rerunFromTask.setPollCount(0);
            if (rerunFromTask.getTaskType().equalsIgnoreCase(TaskType.TASK_TYPE_SUB_WORKFLOW)) {
                // if task is sub workflow set task as IN_PROGRESS and reset start time
                rerunFromTask.setStatus(IN_PROGRESS);
                rerunFromTask.setStartTime(System.currentTimeMillis());
            } else {
                if (taskInput != null) {
                    rerunFromTask.setInputData(taskInput);
                }
                if (systemTaskRegistry.isSystemTask(rerunFromTask.getTaskType())
                        && !systemTaskRegistry.get(rerunFromTask.getTaskType()).isAsync()) {
                    // Start the synchronous system task directly
                    systemTaskRegistry
                            .get(rerunFromTask.getTaskType())
                            .start(workflow, rerunFromTask, this);
                } else if (TaskType.FORK_JOIN_DYNAMIC
                        .name()
                        .equalsIgnoreCase(rerunFromTask.getTaskType())) {
                    // FORK_JOIN_DYNAMIC is not in the system task registry and has no queue worker.
                    // Mark it COMPLETED with executed=false so that decide() re-fires getNextTask()
                    // via ForkJoinDynamicTaskMapper, which re-creates the branch tasks from the
                    // original prep task's output stored in the workflow.
                    rerunFromTask.setStatus(COMPLETED);
                } else {
                    // Set the task to rerun as SCHEDULED
                    rerunFromTask.setStatus(SCHEDULED);
                }
            }
            // Write the new state to DB before queueing so any async worker sees SCHEDULED,
            // not the stale CANCELED/FAILED state that was in the DB before this rerun.
            executionDAOFacade.updateTask(rerunFromTask);
            if (rerunFromTask.getStatus() == SCHEDULED) {
                addTaskToQueue(rerunFromTask);
            }
            if (rerunFromTask.getTaskId().equals(taskId)) {
                // Direct rerun: reset container tasks (DO_WHILE, JOIN) that stayed terminal
                // after seq-based removal, then push parents.
                finalizeRerun(workflow, rerunFromTask);
            } else {
                // Recursive rerun: child workflow already reran; just decide and push parents.
                decide(workflow.getWorkflowId());
                updateAndPushParents(workflow, "reran");
            }
            return true;
        }
        return false;
    }

    private void finalizeRerun(WorkflowModel workflow, TaskModel rerunFromTask) {
        // FORK_JOIN_DYNAMIC rerun: rerunFromTask is the TASK_TYPE_FORK model created by the
        // mapper (task type "FORK", not "FORK_JOIN_DYNAMIC"). Seq-based removal already stripped
        // the branch tasks and JOIN but left the FORK task itself (same seq). Remove it and
        // directly re-invoke the mapper so that branch tasks are re-created — decide()'s
        // getNextTask(FORK) path only returns the JOIN task and never re-expands branches.
        if (TaskType.TASK_TYPE_FORK.equalsIgnoreCase(rerunFromTask.getTaskType())
                && rerunFromTask.getWorkflowTask() != null
                && TaskType.FORK_JOIN_DYNAMIC
                        .name()
                        .equalsIgnoreCase(rerunFromTask.getWorkflowTask().getType())) {
            executionDAOFacade.removeTask(rerunFromTask.getTaskId());
            final String forkTaskId = rerunFromTask.getTaskId();
            workflow.setTasks(
                    workflow.getTasks().stream()
                            .filter(t -> !t.getTaskId().equals(forkTaskId))
                            .collect(Collectors.toList()));
            List<TaskModel> newTasks =
                    deciderService.getTasksToBeScheduled(
                            workflow, rerunFromTask.getWorkflowTask(), 0);
            dedupAndAddTasks(workflow, newTasks);
            scheduleTask(workflow, newTasks);
            queueDAO.push(
                    DECIDER_QUEUE,
                    workflow.getWorkflowId(),
                    workflow.getPriority(),
                    properties.getWorkflowOffsetTimeout().getSeconds());
            decide(workflow.getWorkflowId());
            updateAndPushParents(workflow, "reran");
            return;
        }
        List<TaskModel> tasksToQueue = new ArrayList<>();
        workflow.getTasks()
                .forEach(
                        task -> {
                            if (!task.getStatus().isSuccessful()
                                    && task.getStatus().isTerminal()
                                    && !rerunFromTask
                                            .getReferenceTaskName()
                                            .equals(task.getReferenceTaskName())) {
                                if (TaskType.TASK_TYPE_SUB_WORKFLOW.equalsIgnoreCase(
                                        task.getTaskType())) {
                                    task.setSubWorkflowId(null);
                                    task.getOutputData().remove("subWorkflowId");
                                }
                                if (TaskType.JOIN.toString().equalsIgnoreCase(task.getTaskType())
                                        || TaskType.DO_WHILE
                                                .toString()
                                                .equalsIgnoreCase(task.getTaskType())) {
                                    task.setStatus(IN_PROGRESS);
                                } else if (systemTaskRegistry.isSystemTask(task.getTaskType())
                                        && systemTaskRegistry.get(task.getTaskType()) != null
                                        && !systemTaskRegistry.get(task.getTaskType()).isAsync()) {
                                    task.setStatus(IN_PROGRESS);
                                } else {
                                    task.setStatus(SCHEDULED);
                                    tasksToQueue.add(task);
                                }
                                task.setExecuted(false);
                                task.setStartTime(System.currentTimeMillis());
                                task.setEndTime(0);
                                task.setReasonForIncompletion(null);
                            }
                        });
        // Write SCHEDULED to DB before queueing so async workers (e.g. SystemTaskWorker)
        // never read a stale CANCELED/FAILED state and silently drop the queue entry.
        executionDAOFacade.updateTasks(workflow.getTasks());
        tasksToQueue.forEach(this::addTaskToQueue);
        // Push AFTER all sibling tasks are reset so async decider never sees stale CANCELED/FAILED
        queueDAO.push(
                DECIDER_QUEUE,
                workflow.getWorkflowId(),
                workflow.getPriority(),
                properties.getWorkflowOffsetTimeout().getSeconds());
        decide(workflow.getWorkflowId());
        updateAndPushParents(workflow, "reran");
    }

    @Override
    public void scheduleNextIteration(TaskModel loopTask, WorkflowModel workflow) {
        // Schedule only first loop over task. Rest will be taken care in Decider
        // Service when this
        // task will get completed.
        List<TaskModel> scheduledLoopOverTasks =
                deciderService.getTasksToBeScheduled(
                        workflow,
                        loopTask.getWorkflowTask().getLoopOver().get(0),
                        loopTask.getRetryCount(),
                        null);
        setTaskDomains(scheduledLoopOverTasks, workflow);
        scheduledLoopOverTasks.forEach(
                t -> {
                    t.setReferenceTaskName(
                            TaskUtils.appendIteration(
                                    t.getReferenceTaskName(), loopTask.getIteration()));
                    t.setIteration(loopTask.getIteration());
                });
        scheduleTask(workflow, scheduledLoopOverTasks);
        workflow.getTasks().addAll(scheduledLoopOverTasks);
    }

    private TaskDef getTaskDefinition(TaskModel task) {
        return task.getTaskDefinition()
                .orElseGet(
                        () ->
                                Optional.ofNullable(
                                                metadataDAO.getTaskDef(
                                                        task.getWorkflowTask().getName()))
                                        .orElseThrow(
                                                () -> {
                                                    String reason =
                                                            String.format(
                                                                    "Invalid task specified. Cannot find task by name %s in the task definitions",
                                                                    task.getWorkflowTask()
                                                                            .getName());
                                                    return new TerminateWorkflowException(reason);
                                                }));
    }

    @VisibleForTesting
    void updateParentWorkflowTask(WorkflowModel subWorkflow) {
        TaskModel subWorkflowTask =
                executionDAOFacade.getTaskModel(subWorkflow.getParentWorkflowTaskId());
        if (subWorkflowTask == null) {
            // orphan sub-workflow: parent task was cleared (e.g. parent workflow restarted)
            return;
        }
        executeSubworkflowTaskAndSyncData(subWorkflow, subWorkflowTask);
        executionDAOFacade.updateTask(subWorkflowTask);
        if (subWorkflowTask.getStatus().isTerminal()) {
            // This fork branch's sub-workflow just finished; a sibling JOIN waiting on it may now
            // be satisfiable. Nudge it off its exponential-backoff poll so the parent completes
            // promptly instead of stalling until the JOIN's next scheduled evaluation.
            expediteInProgressJoinTasks(subWorkflowTask.getWorkflowInstanceId());
        }
    }

    private void executeSubworkflowTaskAndSyncData(
            WorkflowModel subWorkflow, TaskModel subWorkflowTask) {
        WorkflowSystemTask subWorkflowSystemTask =
                systemTaskRegistry.get(TaskType.TASK_TYPE_SUB_WORKFLOW);
        subWorkflowSystemTask.execute(subWorkflow, subWorkflowTask, this);
    }

    /**
     * Pushes workflow id into the decider queue with a higher priority to expedite evaluation.
     *
     * @param workflowId The workflow to be evaluated at higher priority
     */
    private void expediteLazyWorkflowEvaluation(String workflowId) {
        if (queueDAO.containsMessage(DECIDER_QUEUE, workflowId)) {
            queueDAO.postpone(DECIDER_QUEUE, workflowId, EXPEDITED_PRIORITY, 0);
        } else {
            queueDAO.push(DECIDER_QUEUE, workflowId, EXPEDITED_PRIORITY, 0);
        }

        LOGGER.info("Pushed workflow {} to {} for expedited evaluation", workflowId, DECIDER_QUEUE);
    }

    private static boolean isJoinOnFailedPermissive(List<String> joinOn, WorkflowModel workflow) {
        return joinOn.stream()
                .map(workflow::getTaskByRefName)
                .anyMatch(
                        t ->
                                t.getWorkflowTask().isPermissive()
                                        && !t.getWorkflowTask().isOptional()
                                        && t.getStatus().equals(FAILED));
    }

    @Override
    public String startWorkflow(StartWorkflowInput input) {
        WorkflowDef workflowDefinition = resolveWorkflowDefinition(input);
        String workflowId =
                Optional.ofNullable(input.getWorkflowId()).orElseGet(idGenerator::generate);
        WorkflowModel workflow = createWorkflowModel(input, workflowDefinition, workflowId);

        try {
            createAndEvaluate(workflow);
            Monitors.recordWorkflowStartSuccess(
                    workflow.getWorkflowName(),
                    String.valueOf(workflow.getWorkflowVersion()),
                    workflow.getOwnerApp());
            return workflowId;
        } catch (Exception e) {
            Monitors.recordWorkflowStartError(
                    workflowDefinition.getName(), WorkflowContext.get().getClientApp());
            LOGGER.error("Unable to start workflow: {}", workflowDefinition.getName(), e);

            // It's possible the remove workflow call hits an exception as well, in that
            // case we
            // want to log both errors to help diagnosis.
            try {
                executionDAOFacade.removeWorkflow(workflowId, false);
            } catch (Exception rwe) {
                LOGGER.error("Could not remove the workflowId: " + workflowId, rwe);
            }
            throw e;
        }
    }

    @Override
    public WorkflowModel startWorkflowIdempotent(StartWorkflowInput input) {
        Preconditions.checkArgument(
                StringUtils.isNotBlank(input.getWorkflowId()),
                "workflowId must be present for idempotent workflow start");

        WorkflowDef workflowDefinition = resolveWorkflowDefinition(input);
        String workflowId = input.getWorkflowId();

        if (!executionLockService.acquireLock(workflowId)) {
            throw new TransientException("Error acquiring lock when creating workflow: {}");
        }

        boolean createAttempted = false;
        try {
            try {
                WorkflowModel existingWorkflow =
                        executionDAOFacade.getWorkflowModelFromExecutionDAO(workflowId, false);
                validateIdempotentWorkflowOwnership(input, existingWorkflow);
                return existingWorkflow;
            } catch (NotFoundException e) {
                LOGGER.debug(
                        "No existing workflow found in execution store for idempotent start of workflow id {}, proceeding with creation",
                        workflowId);
            }

            WorkflowModel workflow = createWorkflowModel(input, workflowDefinition, workflowId);
            createAttempted = true;
            createAndQueueEvaluationWithLock(workflow);

            Monitors.recordWorkflowStartSuccess(
                    workflow.getWorkflowName(),
                    String.valueOf(workflow.getWorkflowVersion()),
                    workflow.getOwnerApp());
            return workflow;
        } catch (Exception e) {
            Monitors.recordWorkflowStartError(
                    workflowDefinition.getName(), WorkflowContext.get().getClientApp());
            LOGGER.error(
                    "Unable to start workflow idempotently: {}", workflowDefinition.getName(), e);

            try {
                if (createAttempted) {
                    executionDAOFacade.removeWorkflow(workflowId, false);
                }
            } catch (Exception rwe) {
                LOGGER.error("Could not remove the workflowId: " + workflowId, rwe);
            }
            throw e;
        } finally {
            executionLockService.releaseLock(workflowId);
        }
    }

    private void validateIdempotentWorkflowOwnership(
            StartWorkflowInput input, WorkflowModel existingWorkflow) {
        if (StringUtils.isBlank(input.getParentWorkflowId())
                && StringUtils.isBlank(input.getParentWorkflowTaskId())) {
            return;
        }

        if (!StringUtils.equals(input.getParentWorkflowId(), existingWorkflow.getParentWorkflowId())
                || !StringUtils.equals(
                        input.getParentWorkflowTaskId(),
                        existingWorkflow.getParentWorkflowTaskId())) {
            String message =
                    String.format(
                            "Workflow id %s already belongs to parent workflow %s task %s, cannot attach to parent workflow %s task %s",
                            existingWorkflow.getWorkflowId(),
                            existingWorkflow.getParentWorkflowId(),
                            existingWorkflow.getParentWorkflowTaskId(),
                            input.getParentWorkflowId(),
                            input.getParentWorkflowTaskId());
            throw new NonTransientException(message);
        }
    }

    private void createAndEvaluate(WorkflowModel workflow) {
        if (!executionLockService.acquireLock(workflow.getWorkflowId())) {
            throw new TransientException("Error acquiring lock when creating workflow: {}");
        }
        try {
            createAndEvaluateWithLock(workflow);
        } finally {
            executionLockService.releaseLock(workflow.getWorkflowId());
        }
    }

    private void createAndEvaluateWithLock(WorkflowModel workflow) {
        executionDAOFacade.createWorkflow(workflow);
        LOGGER.debug(
                "A new instance of workflow: {} created with id: {}",
                workflow.getWorkflowName(),
                workflow.getWorkflowId());
        executionDAOFacade.populateWorkflowAndTaskPayloadData(workflow);
        notifyWorkflowStatusListener(workflow, WorkflowEventType.STARTED);
        decide(workflow);
    }

    private void createAndQueueEvaluationWithLock(WorkflowModel workflow) {
        executionDAOFacade.createWorkflow(workflow);
        LOGGER.debug(
                "A new instance of workflow: {} created with id: {}",
                workflow.getWorkflowName(),
                workflow.getWorkflowId());
        executionDAOFacade.populateWorkflowAndTaskPayloadData(workflow);
        notifyWorkflowStatusListener(workflow, WorkflowEventType.STARTED);
        try {
            expediteLazyWorkflowEvaluation(workflow.getWorkflowId());
        } catch (Exception e) {
            LOGGER.warn(
                    "Unable to expedite evaluation for newly created workflow {}, leaving default decider queue entry in place",
                    workflow.getWorkflowId(),
                    e);
        }
    }

    private WorkflowDef resolveWorkflowDefinition(StartWorkflowInput input) {
        WorkflowDef workflowDefinition;

        if (input.getWorkflowDefinition() == null) {
            workflowDefinition =
                    metadataMapperService.lookupForWorkflowDefinition(
                            input.getName(), input.getVersion());
        } else {
            workflowDefinition = input.getWorkflowDefinition();
        }

        workflowDefinition = metadataMapperService.populateTaskDefinitions(workflowDefinition);
        validateWorkflow(
                workflowDefinition,
                input.getWorkflowInput(),
                input.getExternalInputPayloadStoragePath());
        return workflowDefinition;
    }

    private WorkflowModel createWorkflowModel(
            StartWorkflowInput input, WorkflowDef workflowDefinition, String workflowId) {
        WorkflowModel workflow = new WorkflowModel();
        workflow.setWorkflowId(workflowId);
        workflow.setCorrelationId(input.getCorrelationId());
        workflow.setPriority(input.getPriority() == null ? 0 : input.getPriority());
        workflow.setWorkflowDefinition(workflowDefinition);
        workflow.setStatus(WorkflowModel.Status.RUNNING);
        workflow.setParentWorkflowId(input.getParentWorkflowId());
        workflow.setParentWorkflowTaskId(input.getParentWorkflowTaskId());
        workflow.setOwnerApp(WorkflowContext.get().getClientApp());
        workflow.setCreateTime(System.currentTimeMillis());
        workflow.setUpdatedBy(null);
        workflow.setUpdatedTime(null);
        workflow.setEvent(input.getEvent());
        workflow.setTaskToDomain(input.getTaskToDomain());
        workflow.setVariables(workflowDefinition.getVariables());

        Map<String, Object> workflowInput = input.getWorkflowInput();
        if (workflowInput != null && !workflowInput.isEmpty()) {
            Map<String, Object> parsedInput =
                    parametersUtils.getWorkflowInput(workflowDefinition, workflowInput);
            workflow.setInput(parsedInput);
        } else {
            workflow.setExternalInputPayloadStoragePath(input.getExternalInputPayloadStoragePath());
        }
        return workflow;
    }

    /**
     * Performs validations for starting a workflow
     *
     * @throws IllegalArgumentException if the validation fails.
     */
    private void validateWorkflow(
            WorkflowDef workflowDef,
            Map<String, Object> workflowInput,
            String externalStoragePath) {
        // Check if the input to the workflow is not null
        if (workflowInput == null && StringUtils.isBlank(externalStoragePath)) {
            LOGGER.error("The input for the workflow '{}' cannot be NULL", workflowDef.getName());
            Monitors.recordWorkflowStartError(
                    workflowDef.getName(), WorkflowContext.get().getClientApp());

            throw new IllegalArgumentException("NULL input passed when starting workflow");
        }
    }

    private void notifyWorkflowStatusListener(WorkflowModel workflow, WorkflowEventType event) {
        try {
            switch (event) {
                case STARTED:
                    workflowStatusListener.onWorkflowStartedIfEnabled(workflow);
                    break;
                case RERAN:
                    workflowStatusListener.onWorkflowRerunIfEnabled(workflow);
                    break;
                case RETRIED:
                    workflowStatusListener.onWorkflowRetriedIfEnabled(workflow);
                    break;
                case PAUSED:
                    workflowStatusListener.onWorkflowPausedIfEnabled(workflow);
                    break;
                case RESUMED:
                    workflowStatusListener.onWorkflowResumedIfEnabled(workflow);
                    break;
                case RESTARTED:
                    workflowStatusListener.onWorkflowRestartedIfEnabled(workflow);
                    break;
                case COMPLETED:
                    workflowStatusListener.onWorkflowCompletedIfEnabled(workflow);
                    break;
                case TERMINATED:
                    workflowStatusListener.onWorkflowTerminatedIfEnabled(workflow);
                    break;
                case FINALIZED:
                    workflowStatusListener.onWorkflowFinalizedIfEnabled(workflow);
                    break;
                default:
                    return;
            }
        } catch (Exception e) {
            LOGGER.error(
                    "Error while notifying WorkflowStatusListener for workflow: {}",
                    workflow.getWorkflowId(),
                    e);
        }
    }
}
