/*
 * 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 io.conductor.e2e.control;

import java.util.Arrays;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang3.RandomStringUtils;
import org.junit.jupiter.api.Test;

import com.netflix.conductor.client.http.MetadataClient;
import com.netflix.conductor.client.http.TaskClient;
import com.netflix.conductor.client.http.WorkflowClient;
import com.netflix.conductor.common.metadata.tasks.Task;
import com.netflix.conductor.common.metadata.tasks.TaskDef;
import com.netflix.conductor.common.metadata.tasks.TaskResult;
import com.netflix.conductor.common.metadata.tasks.TaskType;
import com.netflix.conductor.common.metadata.workflow.StartWorkflowRequest;
import com.netflix.conductor.common.metadata.workflow.SubWorkflowParams;
import com.netflix.conductor.common.metadata.workflow.WorkflowDef;
import com.netflix.conductor.common.metadata.workflow.WorkflowTask;
import com.netflix.conductor.common.run.Workflow;

import io.conductor.e2e.util.ApiUtil;
import lombok.extern.slf4j.Slf4j;

import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;

@Slf4j
public class DoWhileWithDomainTests {

    @Test
    public void testSubWorkflow0version() {
        WorkflowClient workflowClient = ApiUtil.WORKFLOW_CLIENT;
        MetadataClient metadataClient = ApiUtil.METADATA_CLIENT;
        TaskClient taskClient = ApiUtil.TASK_CLIENT;

        String parentWorkflowName = RandomStringUtils.randomAlphanumeric(5).toUpperCase();

        // Register workflow
        registerInlineWorkflowDef(parentWorkflowName, metadataClient);

        // Trigger workflow
        StartWorkflowRequest startWorkflowRequest = new StartWorkflowRequest();
        startWorkflowRequest.setName(parentWorkflowName);
        startWorkflowRequest.setVersion(1);

        String workflowId = workflowClient.startWorkflow(startWorkflowRequest);
        log.info("workflowId: {}", workflowId);
        System.out.println("workflowId : " + workflowId);

        await().atMost(33, TimeUnit.SECONDS)
                .untilAsserted(
                        () -> {
                            assertTrue(
                                    !workflowClient
                                            .getWorkflow(workflowId, true)
                                            .getTasks()
                                            .isEmpty());
                        });
        // User1 should be able to complete task/workflow
        String taskRefName =
                workflowClient
                        .getWorkflow(workflowId, true)
                        .getTasks()
                        .get(0)
                        .getReferenceTaskName();
        taskClient.updateTaskSync(workflowId, taskRefName, TaskResult.Status.COMPLETED, Map.of());

        final String[] subWorkflowId = new String[1];
        await().atMost(33, TimeUnit.SECONDS)
                .pollInterval(1, TimeUnit.SECONDS)
                .untilAsserted(
                        () -> {
                            Workflow latestWorkflow = workflowClient.getWorkflow(workflowId, true);
                            assertEquals(
                                    latestWorkflow.getStatus().name(),
                                    Workflow.WorkflowStatus.RUNNING.name());
                            assertTrue(latestWorkflow.getTasks().size() > 1);
                            assertEquals(
                                    latestWorkflow.getTasks().get(0).getStatus(),
                                    Task.Status.COMPLETED);
                            assertEquals(
                                    latestWorkflow.getTasks().get(1).getStatus(),
                                    Task.Status.IN_PROGRESS);
                            subWorkflowId[0] = latestWorkflow.getTasks().get(1).getSubWorkflowId();
                            assertTrue(subWorkflowId[0] != null && !subWorkflowId[0].isBlank());
                        });

        TaskResult taskResult = new TaskResult();
        taskResult.setWorkflowInstanceId(subWorkflowId[0]);
        taskResult.setStatus(TaskResult.Status.COMPLETED);
        taskResult.setTaskId(awaitFirstTaskId(workflowClient, subWorkflowId[0]));
        taskClient.updateTask(taskResult);

        // Wait for workflow to get completed
        await().atMost(33, TimeUnit.SECONDS)
                .untilAsserted(
                        () -> {
                            Workflow workflow1 = workflowClient.getWorkflow(workflowId, true);
                            assertEquals(
                                    workflow1.getTasks().get(0).getStatus(), Task.Status.COMPLETED);
                            assertEquals(
                                    workflow1.getTasks().get(1).getStatus(), Task.Status.COMPLETED);
                        });

        // Cleanup
        metadataClient.unregisterWorkflowDef(parentWorkflowName, 1);
    }

    private String awaitFirstTaskId(WorkflowClient workflowClient, String workflowId) {
        final String[] taskId = new String[1];
        await().atMost(33, TimeUnit.SECONDS)
                .pollInterval(1, TimeUnit.SECONDS)
                .untilAsserted(
                        () -> {
                            Workflow workflow = workflowClient.getWorkflow(workflowId, true);
                            assertTrue(!workflow.getTasks().isEmpty());
                            taskId[0] = workflow.getTasks().get(0).getTaskId();
                            assertTrue(taskId[0] != null && !taskId[0].isBlank());
                        });
        return taskId[0];
    }

    private void registerInlineWorkflowDef(String workflowName, MetadataClient metadataClient1) {
        TaskDef taskDef = new TaskDef("dt1");
        taskDef.setOwnerEmail("test@conductor.io");

        TaskDef taskDef2 = new TaskDef("dt2");
        taskDef2.setOwnerEmail("test@conductor.io");

        WorkflowTask workflowTask = new WorkflowTask();
        workflowTask.setTaskReferenceName("dt2");
        workflowTask.setName("dt2");
        workflowTask.setTaskDefinition(taskDef2);
        workflowTask.setWorkflowTaskType(TaskType.SIMPLE);

        WorkflowTask inline = new WorkflowTask();
        inline.setTaskReferenceName("dt1");
        inline.setName("dt1");
        inline.setTaskDefinition(taskDef);
        inline.setWorkflowTaskType(TaskType.SIMPLE);

        WorkflowTask inlineSubworkflow = new WorkflowTask();
        inlineSubworkflow.setTaskReferenceName("dynamicFork");
        inlineSubworkflow.setName("dynamicFork");
        inlineSubworkflow.setTaskDefinition(taskDef);
        inlineSubworkflow.setWorkflowTaskType(TaskType.SUB_WORKFLOW);

        WorkflowDef inlineWorkflowDef = new WorkflowDef();
        inlineWorkflowDef.setName("inline_test_sub_workflow");
        inlineWorkflowDef.setVersion(1);
        inlineWorkflowDef.setTimeoutSeconds(600);
        inlineWorkflowDef.setTimeoutPolicy(WorkflowDef.TimeoutPolicy.TIME_OUT_WF);
        inlineWorkflowDef.setTasks(Arrays.asList(inline));
        SubWorkflowParams subWorkflowParams = new SubWorkflowParams();
        subWorkflowParams.setName("inline_test_sub_workflow");
        subWorkflowParams.setVersion(1);
        subWorkflowParams.setWorkflowDef(inlineWorkflowDef);
        inlineSubworkflow.setSubWorkflowParam(subWorkflowParams);

        WorkflowDef workflowDef = new WorkflowDef();
        workflowDef.setName(workflowName);
        workflowDef.setOwnerEmail("test@conductor.io");
        workflowDef.setTimeoutSeconds(600);
        workflowDef.setTimeoutPolicy(WorkflowDef.TimeoutPolicy.TIME_OUT_WF);
        workflowDef.setInputParameters(Arrays.asList("value", "inlineValue"));
        workflowDef.setDescription("Workflow to test inline sub_workflow definition");
        workflowDef.setTasks(Arrays.asList(workflowTask, inlineSubworkflow));
        try {
            metadataClient1.updateWorkflowDefs(java.util.List.of(workflowDef));
            metadataClient1.registerTaskDefs(Arrays.asList(taskDef, taskDef2));
        } catch (Exception e) {
        }
    }
}
