/**
*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you 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
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* 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 org.apache.airavata.orchestrator.cpi.impl;

import java.io.File;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.ExecutorService;
import org.apache.airavata.common.exception.AiravataException;
import org.apache.airavata.common.exception.ApplicationSettingsException;
import org.apache.airavata.common.utils.AiravataUtils;
import org.apache.airavata.common.utils.ServerSettings;
import org.apache.airavata.common.utils.ThriftUtils;
import org.apache.airavata.model.appcatalog.appinterface.ApplicationInterfaceDescription;
import org.apache.airavata.model.appcatalog.computeresource.ComputeResourceDescription;
import org.apache.airavata.model.appcatalog.computeresource.JobSubmissionInterface;
import org.apache.airavata.model.appcatalog.computeresource.JobSubmissionProtocol;
import org.apache.airavata.model.appcatalog.computeresource.MonitorMode;
import org.apache.airavata.model.appcatalog.computeresource.SSHJobSubmission;
import org.apache.airavata.model.appcatalog.groupresourceprofile.GroupComputeResourcePreference;
import org.apache.airavata.model.appcatalog.groupresourceprofile.ResourceType;
import org.apache.airavata.model.application.io.DataType;
import org.apache.airavata.model.application.io.InputDataObjectType;
import org.apache.airavata.model.application.io.OutputDataObjectType;
import org.apache.airavata.model.commons.ErrorModel;
import org.apache.airavata.model.data.movement.DataMovementProtocol;
import org.apache.airavata.model.error.LaunchValidationException;
import org.apache.airavata.model.error.ValidationResults;
import org.apache.airavata.model.error.ValidatorResult;
import org.apache.airavata.model.experiment.ExperimentModel;
import org.apache.airavata.model.process.ProcessModel;
import org.apache.airavata.model.scheduling.ComputationalResourceSchedulingModel;
import org.apache.airavata.model.status.TaskState;
import org.apache.airavata.model.status.TaskStatus;
import org.apache.airavata.model.task.DataStageType;
import org.apache.airavata.model.task.DataStagingTaskModel;
import org.apache.airavata.model.task.EnvironmentSetupTaskModel;
import org.apache.airavata.model.task.JobSubmissionTaskModel;
import org.apache.airavata.model.task.MonitorTaskModel;
import org.apache.airavata.model.task.TaskModel;
import org.apache.airavata.model.task.TaskTypes;
import org.apache.airavata.model.util.ExperimentModelUtil;
import org.apache.airavata.orchestrator.core.exception.OrchestratorException;
import org.apache.airavata.orchestrator.core.impl.GFACPassiveJobSubmitter;
import org.apache.airavata.orchestrator.core.job.JobSubmitter;
import org.apache.airavata.orchestrator.core.utils.OrchestratorConstants;
import org.apache.airavata.orchestrator.core.utils.OrchestratorUtils;
import org.apache.airavata.orchestrator.core.validator.JobMetadataValidator;
import org.apache.airavata.registry.api.RegistryService;
import org.apache.airavata.registry.api.client.RegistryServiceClientFactory;
import org.apache.airavata.registry.api.exception.RegistryServiceException;
import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class SimpleOrchestratorImpl extends AbstractOrchestrator {
    private static final Logger logger = LoggerFactory.getLogger(SimpleOrchestratorImpl.class);
    private ExecutorService executor;

    // this is going to be null unless the thread count is 0
    private JobSubmitter jobSubmitter = null;

    public SimpleOrchestratorImpl() throws OrchestratorException, TException {
        try {
            try {
                // We are only going to use GFacPassiveJobSubmitter
                jobSubmitter = new GFACPassiveJobSubmitter();
                jobSubmitter.initialize(this.orchestratorContext);

            } catch (Exception e) {
                String error = "Error creating JobSubmitter in non threaded mode ";
                logger.error(error);
                throw new OrchestratorException(error, e);
            }
        } catch (OrchestratorException e) {
            logger.error("Error Constructing the Orchestrator");
            throw e;
        }
    }

    public boolean launchProcess(ProcessModel processModel, String tokenId) throws OrchestratorException {
        try {
            return jobSubmitter.submit(processModel.getExperimentId(), processModel.getProcessId(), tokenId);
        } catch (Exception e) {
            throw new OrchestratorException("Error launching the job", e);
        }
    }

    public ValidationResults validateExperiment(ExperimentModel experiment)
            throws OrchestratorException, LaunchValidationException {
        ValidationResults validationResults = new ValidationResults();
        validationResults.setValidationState(
                true); // initially making it to success, if atleast one failed them simply mark it failed.
        String errorMsg = "Validation Errors : ";
        if (this.orchestratorConfiguration.isEnableValidation()) {
            List<String> validatorClasses =
                    this.orchestratorContext.getOrchestratorConfiguration().getValidatorClasses();
            for (String validator : validatorClasses) {
                try {
                    Class<? extends JobMetadataValidator> vClass =
                            Class.forName(validator.trim()).asSubclass(JobMetadataValidator.class);
                    JobMetadataValidator jobMetadataValidator = vClass.newInstance();
                    validationResults = jobMetadataValidator.validate(experiment, null);
                    if (validationResults.isValidationState()) {
                        logger.info("Validation of " + validator + " is SUCCESSFUL");
                    } else {
                        List<ValidatorResult> validationResultList = validationResults.getValidationResultList();
                        for (ValidatorResult result : validationResultList) {
                            if (!result.isResult()) {
                                String validationError = result.getErrorDetails();
                                if (validationError != null) {
                                    errorMsg += validationError + " ";
                                }
                            }
                        }
                        logger.error("Validation of " + validator + " for experiment Id " + experiment.getExperimentId()
                                + " is FAILED:[error]. " + errorMsg);
                        validationResults.setValidationState(false);
                        try {
                            ErrorModel details = new ErrorModel();
                            details.setActualErrorMessage(errorMsg);
                            details.setCreationTime(Calendar.getInstance().getTimeInMillis());
                            final RegistryService.Client registryClient = getRegistryServiceClient();
                            try {
                                registryClient.addErrors(
                                        OrchestratorConstants.EXPERIMENT_ERROR, details, experiment.getExperimentId());
                            } finally {
                                if (registryClient != null) {
                                    ThriftUtils.close(registryClient);
                                }
                            }
                        } catch (RegistryServiceException e) {
                            logger.error("Error while saving error details to registry", e);
                            throw new RuntimeException("Error while saving error details to registry", e);
                        } catch (TException e) {
                            throw new RuntimeException("Error while saving error details to registry", e);
                        }
                        break;
                    }
                } catch (ClassNotFoundException e) {
                    logger.error("Error loading the validation class: ", validator, e);
                    validationResults.setValidationState(false);
                } catch (InstantiationException e) {
                    logger.error("Error loading the validation class: ", validator, e);
                    validationResults.setValidationState(false);
                } catch (IllegalAccessException e) {
                    logger.error("Error loading the validation class: ", validator, e);
                    validationResults.setValidationState(false);
                }
            }
        }
        if (validationResults.isValidationState()) {
            return validationResults;
        } else {
            // atleast one validation has failed, so we throw an exception
            LaunchValidationException launchValidationException = new LaunchValidationException();
            launchValidationException.setValidationResult(validationResults);
            launchValidationException.setErrorMessage("Validation failed refer the validationResults list for "
                    + "detail error. Validation errors : " + errorMsg);
            throw launchValidationException;
        }
    }

    public ValidationResults validateProcess(ExperimentModel experiment, ProcessModel processModel)
            throws OrchestratorException, LaunchValidationException {

        ValidationResults validationResults = new ValidationResults();
        validationResults.setValidationState(
                true); // initially making it to success, if atleast one failed them simply mark it failed.
        String errorMsg = "Validation Errors : ";
        if (this.orchestratorConfiguration.isEnableValidation()) {
            List<String> validatorClzzez =
                    this.orchestratorContext.getOrchestratorConfiguration().getValidatorClasses();
            for (String validator : validatorClzzez) {
                try {
                    Class<? extends JobMetadataValidator> vClass =
                            Class.forName(validator.trim()).asSubclass(JobMetadataValidator.class);
                    JobMetadataValidator jobMetadataValidator = vClass.newInstance();
                    validationResults = jobMetadataValidator.validate(experiment, processModel);
                    if (validationResults.isValidationState()) {
                        logger.info("Validation of " + validator + " is SUCCESSFUL");
                    } else {
                        List<ValidatorResult> validationResultList = validationResults.getValidationResultList();
                        for (ValidatorResult result : validationResultList) {
                            if (!result.isResult()) {
                                String validationError = result.getErrorDetails();
                                if (validationError != null) {
                                    errorMsg += validationError + " ";
                                }
                            }
                        }
                        logger.error("Validation of " + validator + " for experiment Id " + experiment.getExperimentId()
                                + " is FAILED:[error]. " + errorMsg);
                        validationResults.setValidationState(false);
                        try {
                            ErrorModel details = new ErrorModel();
                            details.setActualErrorMessage(errorMsg);
                            details.setCreationTime(Calendar.getInstance().getTimeInMillis());
                            final RegistryService.Client registryClient = getRegistryServiceClient();
                            try {
                                registryClient.addErrors(
                                        OrchestratorConstants.PROCESS_ERROR, details, processModel.getProcessId());
                            } finally {
                                if (registryClient != null) {
                                    ThriftUtils.close(registryClient);
                                }
                            }
                        } catch (RegistryServiceException e) {
                            logger.error("Error while saving error details to registry", e);
                            throw new RuntimeException("Error while saving error details to registry", e);
                        } catch (TException e) {
                            throw new RuntimeException("Error while saving error details to registry", e);
                        }
                        break;
                    }
                } catch (ClassNotFoundException | InstantiationException | IllegalAccessException e) {
                    logger.error("Error loading the validation class: ", validator, e);
                    validationResults.setValidationState(false);
                }
            }
        }
        if (validationResults.isValidationState()) {
            return validationResults;
        } else {
            // atleast one validation has failed, so we throw an exception
            LaunchValidationException launchValidationException = new LaunchValidationException();
            launchValidationException.setValidationResult(validationResults);
            launchValidationException.setErrorMessage("Validation failed refer the validationResults "
                    + "list for detail error. Validation errors : " + errorMsg);
            throw launchValidationException;
        }
    }

    public void cancelExperiment(ExperimentModel experiment, String tokenId) throws OrchestratorException {
        logger.info("Terminating experiment " + experiment.getExperimentId());
        RegistryService.Client registryServiceClient = getRegistryServiceClient();

        try {
            List<String> processIds = registryServiceClient.getProcessIds(experiment.getExperimentId());
            if (processIds != null && processIds.size() > 0) {
                for (String processId : processIds) {
                    logger.info("Terminating process " + processId + " of experiment " + experiment.getExperimentId());
                    jobSubmitter.terminate(experiment.getExperimentId(), processId, tokenId);
                }
            } else {
                logger.warn("No processes found for experiment " + experiment.getExperimentId() + " to cancel");
            }
        } catch (TException e) {
            logger.error("Failed to fetch process ids for experiment " + experiment.getExperimentId(), e);
            throw new OrchestratorException(
                    "Failed to fetch process ids for experiment " + experiment.getExperimentId(), e);
        }
    }

    public ExecutorService getExecutor() {
        return executor;
    }

    public void setExecutor(ExecutorService executor) {
        this.executor = executor;
    }

    public JobSubmitter getJobSubmitter() {
        return jobSubmitter;
    }

    public void setJobSubmitter(JobSubmitter jobSubmitter) {
        this.jobSubmitter = jobSubmitter;
    }

    public void initialize() throws OrchestratorException {}

    public List<ProcessModel> createProcesses(String experimentId, String gatewayId) throws OrchestratorException {
        final RegistryService.Client registryClient = getRegistryServiceClient();
        try {
            ExperimentModel experimentModel = registryClient.getExperiment(experimentId);
            List<ProcessModel> processModels = registryClient.getProcessList(experimentId);
            if (processModels == null || processModels.isEmpty()) {
                ProcessModel processModel = ExperimentModelUtil.cloneProcessFromExperiment(experimentModel);
                String processId = registryClient.addProcess(processModel, experimentId);
                processModel.setProcessId(processId);
                processModels = new ArrayList<>();
                processModels.add(processModel);
            }
            return processModels;
        } catch (Exception e) {
            throw new OrchestratorException("Error during creating process", e);
        } finally {
            if (registryClient != null) {
                ThriftUtils.close(registryClient);
            }
        }
    }

    public String createAndSaveTasks(String gatewayId, ProcessModel processModel) throws OrchestratorException {
        final RegistryService.Client registryClient = getRegistryServiceClient();
        try {
            GroupComputeResourcePreference preference =
                    OrchestratorUtils.getGroupComputeResourcePreference(processModel);
            ResourceType resourceType = preference.getResourceType();
            logger.info("Determined resource type as {} for process {}", resourceType, processModel.getProcessId());

            ComputationalResourceSchedulingModel resourceSchedule = processModel.getProcessResourceSchedule();
            int userGivenWallTime = resourceSchedule.getWallTimeLimit();
            String resourceHostId = resourceSchedule.getResourceHostId();
            if (resourceHostId == null) {
                throw new OrchestratorException("Compute Resource Id cannot be null at this point");
            }

            // TODO - handle for different resource types
            JobSubmissionInterface preferredJobSubmissionInterface =
                    OrchestratorUtils.getPreferredJobSubmissionInterface(processModel, gatewayId);
            JobSubmissionProtocol preferredJobSubmissionProtocol =
                    OrchestratorUtils.getPreferredJobSubmissionProtocol(processModel, gatewayId);
            List<String> taskIdList = new ArrayList<>();

            if (preferredJobSubmissionProtocol == JobSubmissionProtocol.UNICORE) {
                // TODO - breakdown unicore all in one task to multiple tasks, then we don't need to handle UNICORE
                // here.
                taskIdList.addAll(createAndSaveSubmissionTasks(
                        registryClient, preferredJobSubmissionInterface, processModel, userGivenWallTime));
            } else {
                taskIdList.addAll(createAndSaveEnvSetupTask(registryClient, gatewayId, processModel, resourceType));
                taskIdList.addAll(createAndSaveInputDataStagingTasks(processModel, gatewayId, resourceType));
                taskIdList.addAll(createAndSaveSubmissionTasks(
                        registryClient, preferredJobSubmissionInterface, processModel, userGivenWallTime));
                taskIdList.addAll(createAndSaveOutputDataStagingTasks(processModel, gatewayId, resourceType));
            }
            // update process scheduling
            registryClient.updateProcess(processModel, processModel.getProcessId());
            return getTaskDag(taskIdList);
        } catch (Exception e) {
            throw new OrchestratorException("Error during creating process", e);
        } finally {
            if (registryClient != null) {
                ThriftUtils.close(registryClient);
            }
        }
    }

    public String createAndSaveIntermediateOutputFetchingTasks(
            String gatewayId, ProcessModel processModel, ProcessModel parentProcess) throws OrchestratorException {
        final RegistryService.Client registryClient = getRegistryServiceClient();
        try {
            GroupComputeResourcePreference preference =
                    OrchestratorUtils.getGroupComputeResourcePreference(processModel);
            ResourceType resourceType = preference.getResourceType();
            List<String> taskIdList = new ArrayList<>(createAndSaveIntermediateOutputDataStagingTasks(
                    processModel, gatewayId, parentProcess, resourceType));
            // update process scheduling
            registryClient.updateProcess(processModel, processModel.getProcessId());
            return getTaskDag(taskIdList);
        } catch (Exception e) {
            throw new OrchestratorException("Error during creating process", e);
        } finally {
            if (registryClient != null) {
                ThriftUtils.close(registryClient);
            }
        }
    }

    private String getTaskDag(List<String> taskIdList) {
        if (taskIdList.isEmpty()) {
            return "";
        }
        StringBuilder sb = new StringBuilder();
        for (String s : taskIdList) {
            sb.append(s).append(","); // comma separated values
        }
        String dag = sb.toString();
        return dag.substring(0, dag.length() - 1); // remove last comma
    }

    private List<String> createAndSaveEnvSetupTask(
            RegistryService.Client registryClient,
            String gatewayId,
            ProcessModel processModel,
            ResourceType resourceType)
            throws TException, AiravataException, OrchestratorException {
        List<String> envTaskIds = new ArrayList<>();

        TaskModel envSetupTask = new TaskModel();
        envSetupTask.setTaskType(TaskTypes.ENV_SETUP);
        envSetupTask.setTaskStatuses(List.of(new TaskStatus(TaskState.CREATED)));
        envSetupTask.setCreationTime(AiravataUtils.getCurrentTimestamp().getTime());
        envSetupTask.setLastUpdateTime(AiravataUtils.getCurrentTimestamp().getTime());
        envSetupTask.setParentProcessId(processModel.getProcessId());

        EnvironmentSetupTaskModel envSetupSubModel = new EnvironmentSetupTaskModel();
        envSetupSubModel.setProtocol(
                OrchestratorUtils.getSecurityProtocol(processModel, gatewayId)); // TODO support for CLOUD (AWS)

        String scratchLocation = OrchestratorUtils.getScratchLocation(processModel, gatewayId);
        String workingDir = scratchLocation + File.separator + processModel.getProcessId();
        envSetupSubModel.setLocation(workingDir);

        byte[] envSetupSub = ThriftUtils.serializeThriftObject(envSetupSubModel);
        envSetupTask.setSubTaskModel(envSetupSub);
        envSetupTask.setMaxRetry(3);
        envSetupTask.setCurrentRetry(0);
        String envSetupTaskId = registryClient.addTask(envSetupTask, processModel.getProcessId());
        envSetupTask.setTaskId(envSetupTaskId);
        envTaskIds.add(envSetupTaskId);

        return envTaskIds;
    }

    public List<String> createAndSaveInputDataStagingTasks(
            ProcessModel processModel, String gatewayId, ResourceType resourceType) throws AiravataException {

        List<String> dataStagingTaskIds = new ArrayList<>();
        List<InputDataObjectType> processInputs = processModel.getProcessInputs();

        sortByInputOrder(processInputs);
        if (processInputs != null) {
            for (InputDataObjectType processInput : processInputs) {
                DataType type = processInput.getType();
                switch (type) {
                    case STDERR:
                        break;
                    case STDOUT:
                        break;
                    case URI:
                    case URI_COLLECTION:
                        if ((processInput.getValue() == null
                                        || processInput.getValue().isEmpty())
                                && !processInput.isIsRequired()) {
                            logger.debug(
                                    "Skipping input data staging task for {} since value is empty and not required",
                                    processInput.getName());
                            break;
                        }
                        final RegistryService.Client registryClient = getRegistryServiceClient();
                        try {
                            TaskModel inputDataStagingTask = getInputDataStagingTask(
                                    registryClient, processModel, processInput, gatewayId, resourceType);
                            String taskId = registryClient.addTask(inputDataStagingTask, processModel.getProcessId());
                            inputDataStagingTask.setTaskId(taskId);
                            dataStagingTaskIds.add(inputDataStagingTask.getTaskId());
                        } catch (Exception e) {
                            throw new AiravataException("Error while serializing data staging sub task model", e);
                        } finally {
                            if (registryClient != null) {
                                ThriftUtils.close(registryClient);
                            }
                        }
                        break;
                    default:
                        // nothing to do
                        break;
                }
            }
        }
        return dataStagingTaskIds;
    }

    public List<String> createAndSaveOutputDataStagingTasks(
            ProcessModel processModel, String gatewayId, ResourceType resourceType)
            throws AiravataException, TException, OrchestratorException {

        final RegistryService.Client registryClient = getRegistryServiceClient();
        List<String> dataStagingTaskIds = new ArrayList<>();
        try {
            List<OutputDataObjectType> processOutputs = processModel.getProcessOutputs();
            String appName = OrchestratorUtils.getApplicationInterfaceName(processModel);
            if (processOutputs != null) {
                for (OutputDataObjectType processOutput : processOutputs) {
                    DataType type = processOutput.getType();
                    switch (type) {
                        case STDOUT:
                            if (null == processOutput.getValue()
                                    || processOutput.getValue().trim().isEmpty()) {
                                processOutput.setValue(appName + ".stdout");
                            }
                            createOutputDataSatagingTasks(
                                    registryClient,
                                    processModel,
                                    gatewayId,
                                    dataStagingTaskIds,
                                    processOutput,
                                    resourceType);
                            break;
                        case STDERR:
                            if (null == processOutput.getValue()
                                    || processOutput.getValue().trim().isEmpty()) {
                                processOutput.setValue(appName + ".stderr");
                            }
                            createOutputDataSatagingTasks(
                                    registryClient,
                                    processModel,
                                    gatewayId,
                                    dataStagingTaskIds,
                                    processOutput,
                                    resourceType);
                            break;
                        case URI:
                        case URI_COLLECTION:
                            createOutputDataSatagingTasks(
                                    registryClient,
                                    processModel,
                                    gatewayId,
                                    dataStagingTaskIds,
                                    processOutput,
                                    resourceType);
                            break;
                        default:
                            // nothing to do
                            break;
                    }
                }
            }

            try {
                if (isArchive(registryClient, processModel)) {
                    createArchiveDataStatgingTask(
                            registryClient, processModel, gatewayId, dataStagingTaskIds, resourceType);
                }
            } catch (Exception e) {
                throw new AiravataException("Error! Application interface retrieval failed", e);
            }
        } finally {
            if (registryClient != null) {
                ThriftUtils.close(registryClient);
            }
        }
        return dataStagingTaskIds;
    }

    public List<String> createAndSaveIntermediateOutputDataStagingTasks(
            ProcessModel processModel, String gatewayId, ProcessModel parentProcess, ResourceType resourceType)
            throws AiravataException, TException, OrchestratorException {

        final RegistryService.Client registryClient = getRegistryServiceClient();
        List<String> dataStagingTaskIds = new ArrayList<>();
        try {
            List<OutputDataObjectType> processOutputs = processModel.getProcessOutputs();
            String appName = OrchestratorUtils.getApplicationInterfaceName(processModel);
            if (processOutputs != null) {
                for (OutputDataObjectType processOutput : processOutputs) {
                    DataType type = processOutput.getType();
                    switch (type) {
                        case STDOUT:
                            if (null == processOutput.getValue()
                                    || processOutput.getValue().trim().isEmpty()) {
                                processOutput.setValue(appName + ".stdout");
                            }
                            createIntermediateOutputDataStagingTasks(
                                    registryClient,
                                    processModel,
                                    gatewayId,
                                    parentProcess,
                                    dataStagingTaskIds,
                                    processOutput,
                                    resourceType);
                            break;
                        case STDERR:
                            if (null == processOutput.getValue()
                                    || processOutput.getValue().trim().isEmpty()) {
                                processOutput.setValue(appName + ".stderr");
                            }
                            createIntermediateOutputDataStagingTasks(
                                    registryClient,
                                    processModel,
                                    gatewayId,
                                    parentProcess,
                                    dataStagingTaskIds,
                                    processOutput,
                                    resourceType);
                            break;
                        case URI:
                        case URI_COLLECTION:
                            createIntermediateOutputDataStagingTasks(
                                    registryClient,
                                    processModel,
                                    gatewayId,
                                    parentProcess,
                                    dataStagingTaskIds,
                                    processOutput,
                                    resourceType);
                            break;
                        default:
                            // nothing to do
                            break;
                    }
                }
            }

        } finally {
            if (registryClient != null) {
                ThriftUtils.close(registryClient);
            }
        }
        return dataStagingTaskIds;
    }

    private boolean isArchive(RegistryService.Client registryClient, ProcessModel processModel) throws TException {
        ApplicationInterfaceDescription appInterface =
                registryClient.getApplicationInterface(processModel.getApplicationInterfaceId());
        return appInterface.isArchiveWorkingDirectory();
    }

    private void createArchiveDataStatgingTask(
            RegistryService.Client registryClient,
            ProcessModel processModel,
            String gatewayId,
            List<String> dataStagingTaskIds,
            ResourceType resourceType)
            throws AiravataException, TException, OrchestratorException {
        TaskModel archiveTask;
        try {
            archiveTask = getOutputDataStagingTask(registryClient, processModel, null, gatewayId, null, resourceType);
        } catch (TException e) {
            throw new AiravataException("Error! DataStaging sub task serialization failed", e);
        }
        String taskId = registryClient.addTask(archiveTask, processModel.getProcessId());
        archiveTask.setTaskId(taskId);
        dataStagingTaskIds.add(archiveTask.getTaskId());
    }

    private void createOutputDataSatagingTasks(
            RegistryService.Client registryClient,
            ProcessModel processModel,
            String gatewayId,
            List<String> dataStagingTaskIds,
            OutputDataObjectType processOutput,
            ResourceType resourceType)
            throws AiravataException, OrchestratorException {
        try {
            TaskModel outputDataStagingTask = getOutputDataStagingTask(
                    registryClient, processModel, processOutput, gatewayId, null, resourceType);
            String taskId = registryClient.addTask(outputDataStagingTask, processModel.getProcessId());
            outputDataStagingTask.setTaskId(taskId);
            dataStagingTaskIds.add(outputDataStagingTask.getTaskId());
        } catch (TException e) {
            throw new AiravataException("Error while serializing data staging sub task model", e);
        }
    }

    private void createIntermediateOutputDataStagingTasks(
            RegistryService.Client registryClient,
            ProcessModel processModel,
            String gatewayId,
            ProcessModel parentProcess,
            List<String> dataStagingTaskIds,
            OutputDataObjectType processOutput,
            ResourceType resourceType)
            throws AiravataException, OrchestratorException {
        try {
            TaskModel outputDataStagingTask = getOutputDataStagingTask(
                    registryClient, processModel, processOutput, gatewayId, parentProcess, resourceType);
            outputDataStagingTask.setTaskType(TaskTypes.OUTPUT_FETCHING);
            String taskId = registryClient.addTask(outputDataStagingTask, processModel.getProcessId());
            outputDataStagingTask.setTaskId(taskId);
            dataStagingTaskIds.add(outputDataStagingTask.getTaskId());
        } catch (TException e) {
            throw new AiravataException("Error while serializing data staging sub task model", e);
        }
    }

    private List<String> createAndSaveSubmissionTasks(
            RegistryService.Client registryClient,
            JobSubmissionInterface jobSubmissionInterface,
            ProcessModel processModel,
            int wallTime)
            throws TException, OrchestratorException {

        JobSubmissionProtocol jobSubmissionProtocol = jobSubmissionInterface.getJobSubmissionProtocol();
        MonitorMode monitorMode;

        if (jobSubmissionProtocol == JobSubmissionProtocol.SSH
                || jobSubmissionProtocol == JobSubmissionProtocol.SSH_FORK) {
            SSHJobSubmission sshJobSubmission =
                    OrchestratorUtils.getSSHJobSubmission(jobSubmissionInterface.getJobSubmissionInterfaceId());
            monitorMode = sshJobSubmission.getMonitorMode();
        } else if (jobSubmissionProtocol == JobSubmissionProtocol.UNICORE) {
            monitorMode = MonitorMode.FORK;
        } else if (jobSubmissionProtocol == JobSubmissionProtocol.LOCAL) {
            monitorMode = MonitorMode.LOCAL;
        } else if (jobSubmissionProtocol == JobSubmissionProtocol.CLOUD) {
            monitorMode = MonitorMode.CLOUD_JOB_MONITOR;
        } else {
            logger.error(
                    "expId : {}, processId : {} :- Unsupported Job submission protocol {}.",
                    processModel.getExperimentId(),
                    processModel.getProcessId(),
                    jobSubmissionProtocol.name());
            throw new OrchestratorException("Unsupported Job Submission Protocol " + jobSubmissionProtocol.name());
        }

        List<String> submissionTaskIds = new ArrayList<>();
        TaskStatus taskStatus = new TaskStatus(TaskState.CREATED);
        taskStatus.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime());

        JobSubmissionTaskModel submissionSubTask = new JobSubmissionTaskModel();
        submissionSubTask.setMonitorMode(monitorMode);
        submissionSubTask.setJobSubmissionProtocol(jobSubmissionProtocol);
        submissionSubTask.setWallTime(wallTime);

        byte[] bytes = ThriftUtils.serializeThriftObject(submissionSubTask);

        TaskModel taskModel = new TaskModel();
        taskModel.setParentProcessId(processModel.getProcessId());
        taskModel.setCreationTime(System.currentTimeMillis());
        taskModel.setLastUpdateTime(taskModel.getCreationTime());
        taskModel.setTaskStatuses(List.of(taskStatus));
        taskModel.setTaskType(TaskTypes.JOB_SUBMISSION);
        taskModel.setSubTaskModel(bytes);
        taskModel.setMaxRetry(1);
        taskModel.setCurrentRetry(0);

        String taskId = registryClient.addTask(taskModel, processModel.getProcessId());
        taskModel.setTaskId(taskId);
        submissionTaskIds.add(taskModel.getTaskId());

        // create monitor task for this Email based monitor mode job
        if (monitorMode == MonitorMode.JOB_EMAIL_NOTIFICATION_MONITOR || monitorMode == MonitorMode.CLOUD_JOB_MONITOR) {

            TaskStatus monitorTaskStatus = new TaskStatus(TaskState.CREATED);
            monitorTaskStatus.setTimeOfStateChange(
                    AiravataUtils.getCurrentTimestamp().getTime());

            TaskModel monitorTaskModel = new TaskModel();
            monitorTaskModel.setParentProcessId(processModel.getProcessId());
            monitorTaskModel.setCreationTime(System.currentTimeMillis());
            monitorTaskModel.setLastUpdateTime(monitorTaskModel.getCreationTime());
            monitorTaskModel.setTaskStatuses(List.of(monitorTaskStatus));
            monitorTaskModel.setTaskType(TaskTypes.MONITORING);

            MonitorTaskModel monitorSubTaskModel = new MonitorTaskModel();
            monitorSubTaskModel.setMonitorMode(monitorMode);
            monitorTaskModel.setSubTaskModel(ThriftUtils.serializeThriftObject(monitorSubTaskModel));

            String mTaskId = registryClient.addTask(monitorTaskModel, processModel.getProcessId());
            monitorTaskModel.setTaskId(mTaskId);
            submissionTaskIds.add(monitorTaskModel.getTaskId());
        }

        return submissionTaskIds;
    }

    private void sortByInputOrder(List<InputDataObjectType> processInputs) {
        Collections.sort(processInputs, new Comparator<InputDataObjectType>() {
            @Override
            public int compare(InputDataObjectType inputDT_1, InputDataObjectType inputDT_2) {
                return inputDT_1.getInputOrder() - inputDT_2.getInputOrder();
            }
        });
    }

    private TaskModel getInputDataStagingTask(
            RegistryService.Client registryClient,
            ProcessModel processModel,
            InputDataObjectType processInput,
            String gatewayId,
            ResourceType resourceType)
            throws TException, AiravataException, OrchestratorException {
        // create new task model for this task
        TaskModel taskModel = new TaskModel();
        taskModel.setParentProcessId(processModel.getProcessId());
        taskModel.setCreationTime(AiravataUtils.getCurrentTimestamp().getTime());
        taskModel.setLastUpdateTime(taskModel.getCreationTime());
        TaskStatus taskStatus = new TaskStatus(TaskState.CREATED);
        taskStatus.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime());
        taskModel.setTaskStatuses(Arrays.asList(taskStatus));
        taskModel.setTaskType(TaskTypes.DATA_STAGING);
        // create data staging sub task model
        DataStagingTaskModel submodel = new DataStagingTaskModel();
        ComputeResourceDescription computeResource =
                registryClient.getComputeResource(processModel.getComputeResourceId());

        String scratchLocation = OrchestratorUtils.getScratchLocation(processModel, gatewayId);
        String workingDir =
                (scratchLocation.endsWith(File.separator) ? scratchLocation : scratchLocation + File.separator)
                        + processModel.getProcessId()
                        + File.separator;

        URI destination;
        try {
            DataMovementProtocol dataMovementProtocol =
                    OrchestratorUtils.getPreferredDataMovementProtocol(processModel, gatewayId);
            String loginUserName = OrchestratorUtils.getLoginUserName(processModel, gatewayId);
            StringBuilder destinationPath = new StringBuilder(workingDir);
            Optional.ofNullable(processInput.getOverrideFilename())
                    .ifPresent(destinationPath::append); // If an override filename is provided

            destination = new URI(
                    dataMovementProtocol.name(),
                    loginUserName,
                    computeResource.getHostName(),
                    OrchestratorUtils.getDataMovementPort(processModel, gatewayId),
                    destinationPath.toString(),
                    null,
                    null);

        } catch (URISyntaxException e) {
            logger.error("Error while constructing destination file URI", e);
            throw new OrchestratorException("Error while constructing destination file URI", e);
        }

        submodel.setDestination(destination.toString());
        submodel.setType(DataStageType.INPUT);
        submodel.setSource(processInput.getValue());
        submodel.setProcessInput(processInput);
        taskModel.setSubTaskModel(ThriftUtils.serializeThriftObject(submodel));
        taskModel.setMaxRetry(3);
        taskModel.setCurrentRetry(0);
        return taskModel;
    }

    private TaskModel getOutputDataStagingTask(
            RegistryService.Client registryClient,
            ProcessModel processModel,
            OutputDataObjectType processOutput,
            String gatewayId,
            ProcessModel parentProcess,
            ResourceType resourceType)
            throws TException, AiravataException, OrchestratorException {
        try {
            // create new task model for this task
            TaskStatus taskStatus = new TaskStatus(TaskState.CREATED);
            taskStatus.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime());

            TaskModel taskModel = new TaskModel();
            taskModel.setParentProcessId(processModel.getProcessId());
            taskModel.setCreationTime(AiravataUtils.getCurrentTimestamp().getTime());
            taskModel.setLastUpdateTime(taskModel.getCreationTime());
            taskModel.setTaskStatuses(List.of(taskStatus));
            taskModel.setTaskType(TaskTypes.DATA_STAGING);

            ComputeResourceDescription computeResource =
                    registryClient.getComputeResource(processModel.getComputeResourceId());
            DataStagingTaskModel submodel = new DataStagingTaskModel();

            String workingDir = OrchestratorUtils.getScratchLocation(processModel, gatewayId)
                    + File.separator
                    + (parentProcess == null ? processModel.getProcessId() : parentProcess.getProcessId())
                    + File.separator;
            DataMovementProtocol dataMovementProtocol =
                    OrchestratorUtils.getPreferredDataMovementProtocol(processModel, gatewayId);
            URI source;
            try {
                String loginUserName = OrchestratorUtils.getLoginUserName(processModel, gatewayId);
                if (processOutput != null) {
                    submodel.setType(DataStageType.OUPUT);
                    submodel.setProcessOutput(processOutput);
                    source = new URI(
                            dataMovementProtocol.name(),
                            loginUserName,
                            computeResource.getHostName(),
                            OrchestratorUtils.getDataMovementPort(processModel, gatewayId),
                            workingDir + processOutput.getValue(),
                            null,
                            null);
                } else {
                    // archive
                    submodel.setType(DataStageType.ARCHIVE_OUTPUT);
                    source = new URI(
                            dataMovementProtocol.name(),
                            loginUserName,
                            computeResource.getHostName(),
                            OrchestratorUtils.getDataMovementPort(processModel, gatewayId),
                            workingDir,
                            null,
                            null);
                }

            } catch (URISyntaxException e) {
                throw new OrchestratorException("Error while constructing source file URI", e);
            }

            submodel.setSource(source.toString());
            // We don't know destination location at this time, data staging task will set this.
            // because destination is required field we set dummy destination
            submodel.setDestination("dummy://temp/file/location");
            taskModel.setSubTaskModel(ThriftUtils.serializeThriftObject(submodel));
            taskModel.setMaxRetry(3);
            taskModel.setCurrentRetry(0);
            return taskModel;

        } catch (OrchestratorException e) {
            throw new OrchestratorException("Error occurred while retrieving data movement from app catalog", e);
        }
    }

    private RegistryService.Client getRegistryServiceClient() {
        try {
            final int serverPort = Integer.parseInt(ServerSettings.getRegistryServerPort());
            final String serverHost = ServerSettings.getRegistryServerHost();
            return RegistryServiceClientFactory.createRegistryClient(serverHost, serverPort);
        } catch (RegistryServiceException | ApplicationSettingsException e) {
            throw new RuntimeException("Unable to create registry client...", e);
        }
    }
}
