/*
 * Copyright 2025 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 org.conductoross.conductor.core.execution;

import java.time.Duration;
import java.util.Arrays;
import java.util.List;

import org.junit.Test;

import com.netflix.conductor.common.metadata.tasks.TaskDef;
import com.netflix.conductor.common.metadata.tasks.TaskType;
import com.netflix.conductor.common.metadata.workflow.WorkflowDef;
import com.netflix.conductor.common.metadata.workflow.WorkflowTask;
import com.netflix.conductor.model.TaskModel;
import com.netflix.conductor.model.WorkflowModel;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;

public class ExecutorUtilsTest {

    @Test
    public void computePostponeUsesMinimumAcrossTasks() {
        TaskModel scheduledLong = newTask(TaskType.TASK_TYPE_SIMPLE, TaskModel.Status.SCHEDULED);
        TaskDef longPollDef = new TaskDef();
        longPollDef.setPollTimeoutSeconds(50);
        WorkflowTask longWorkflowTask = new WorkflowTask();
        longWorkflowTask.setTaskDefinition(longPollDef);
        scheduledLong.setWorkflowTask(longWorkflowTask);

        TaskModel scheduledShort = newTask(TaskType.TASK_TYPE_SIMPLE, TaskModel.Status.SCHEDULED);
        TaskDef shortPollDef = new TaskDef();
        shortPollDef.setPollTimeoutSeconds(5);
        WorkflowTask shortWorkflowTask = new WorkflowTask();
        shortWorkflowTask.setTaskDefinition(shortPollDef);
        scheduledShort.setWorkflowTask(shortWorkflowTask);

        WorkflowModel workflow =
                newWorkflow(Arrays.asList(scheduledLong, scheduledShort), 0 /* timeoutSeconds */);

        Duration result =
                ExecutorUtils.computePostpone(
                        workflow, Duration.ofSeconds(30), Duration.ofSeconds(3600));

        assertEquals(6, result.getSeconds());
    }

    @Test
    public void computePostponeCapsByMaxPostpone() {
        TaskModel responseTask = newTask(TaskType.TASK_TYPE_SIMPLE, TaskModel.Status.IN_PROGRESS);
        responseTask.setResponseTimeoutSeconds(500);
        responseTask.setStartTime(System.currentTimeMillis());

        WorkflowModel workflow = newWorkflow(Arrays.asList(responseTask), 0);

        Duration result =
                ExecutorUtils.computePostpone(
                        workflow, Duration.ofSeconds(30), Duration.ofSeconds(60));

        assertEquals(60, result.getSeconds());
    }

    @Test
    public void computePostponeUsesScheduledPollTimeout() {
        TaskDef taskDef = new TaskDef();
        taskDef.setPollTimeoutSeconds(10);
        WorkflowTask workflowTask = new WorkflowTask();
        workflowTask.setTaskDefinition(taskDef);

        TaskModel scheduled = newTask(TaskType.TASK_TYPE_SIMPLE, TaskModel.Status.SCHEDULED);
        scheduled.setWorkflowTask(workflowTask);

        WorkflowModel workflow = newWorkflow(Arrays.asList(scheduled), 0);

        Duration result =
                ExecutorUtils.computePostpone(
                        workflow, Duration.ofSeconds(30), Duration.ofSeconds(3600));

        assertEquals(11, result.getSeconds());
    }

    @Test
    public void computePostponeUsesWorkflowOffsetForScheduledSubWorkflow() {
        TaskModel scheduledSubWorkflow =
                newTask(TaskType.TASK_TYPE_SUB_WORKFLOW, TaskModel.Status.SCHEDULED);

        WorkflowModel workflow = newWorkflow(Arrays.asList(scheduledSubWorkflow), 432000);

        Duration result =
                ExecutorUtils.computePostpone(
                        workflow, Duration.ofSeconds(30), Duration.ofSeconds(900));

        assertEquals(30, result.getSeconds());
    }

    @Test
    public void computePostponePrefersScheduledSubWorkflowOffsetOverWorkflowTimeout() {
        TaskModel scheduledSubWorkflow =
                newTask(TaskType.TASK_TYPE_SUB_WORKFLOW, TaskModel.Status.SCHEDULED);
        TaskModel scheduledSimple = newTask(TaskType.TASK_TYPE_SIMPLE, TaskModel.Status.SCHEDULED);

        WorkflowModel workflow =
                newWorkflow(Arrays.asList(scheduledSimple, scheduledSubWorkflow), 432000);

        Duration result =
                ExecutorUtils.computePostpone(
                        workflow, Duration.ofSeconds(30), Duration.ofSeconds(900));

        assertEquals(30, result.getSeconds());
    }

    @Test
    public void computePostponeUsesWorkflowOffsetForInProgressSubWorkflow() {
        TaskModel inProgressSubWorkflow =
                newTask(TaskType.TASK_TYPE_SUB_WORKFLOW, TaskModel.Status.IN_PROGRESS);
        inProgressSubWorkflow.setResponseTimeoutSeconds(500);
        inProgressSubWorkflow.setStartTime(System.currentTimeMillis());

        WorkflowModel workflow = newWorkflow(Arrays.asList(inProgressSubWorkflow), 432000);

        Duration result =
                ExecutorUtils.computePostpone(
                        workflow, Duration.ofSeconds(30), Duration.ofSeconds(900));

        assertEquals(30, result.getSeconds());
    }

    @Test
    public void computePostponeDefaultsToWorkflowOffsetWhenNoEligibleTasks() {
        TaskModel completed = newTask(TaskType.TASK_TYPE_SIMPLE, TaskModel.Status.COMPLETED);
        WorkflowModel workflow = newWorkflow(Arrays.asList(completed), 0);

        Duration result =
                ExecutorUtils.computePostpone(
                        workflow, Duration.ofSeconds(30), Duration.ofSeconds(3600));

        assertEquals(30, result.getSeconds());
    }

    @Test
    public void computePostponeUsesOffsetForWaitWithoutTimeout() {
        TaskModel waitTask = newTask(TaskType.TASK_TYPE_WAIT, TaskModel.Status.IN_PROGRESS);
        waitTask.setWaitTimeout(0);
        WorkflowModel workflow = newWorkflow(Arrays.asList(waitTask), 0);

        Duration result =
                ExecutorUtils.computePostpone(
                        workflow, Duration.ofSeconds(30), Duration.ofSeconds(3600));

        assertEquals(30, result.getSeconds());
    }

    /** Boundary: SCHEDULED task whose poll window has already elapsed → remaining = 0. */
    @Test
    public void computePostponeScheduledElapsedExceedsTimeout() {
        TaskModel task = newTask(TaskType.TASK_TYPE_SIMPLE, TaskModel.Status.SCHEDULED);
        TaskDef taskDef = new TaskDef();
        taskDef.setPollTimeoutSeconds(1); // 1-second poll window
        WorkflowTask workflowTask = new WorkflowTask();
        workflowTask.setTaskDefinition(taskDef);
        task.setWorkflowTask(workflowTask);
        // Push scheduledTime far into the past so elapsed >> pollTimeout
        task.setScheduledTime(System.currentTimeMillis() - 60_000);

        WorkflowModel workflow = newWorkflow(Arrays.asList(task), 0);
        Duration result =
                ExecutorUtils.computePostpone(
                        workflow, Duration.ofSeconds(30), Duration.ofSeconds(3600));

        // Elapsed >> pollTimeout: remaining = 0, clamped to 0
        assertEquals(
                "When poll window is already elapsed, postpone should be 0",
                0L,
                result.getSeconds());
    }

    /** Boundary: IN_PROGRESS task whose response window has already elapsed → remaining = 0. */
    @Test
    public void computePostponeInProgressElapsedExceedsResponseTimeout() {
        TaskModel task = newTask(TaskType.TASK_TYPE_SIMPLE, TaskModel.Status.IN_PROGRESS);
        task.setResponseTimeoutSeconds(1); // 1-second response window
        // Push startTime far into the past so elapsed >> responseTimeout
        task.setStartTime(System.currentTimeMillis() - 60_000);

        WorkflowModel workflow = newWorkflow(Arrays.asList(task), 0);
        Duration result =
                ExecutorUtils.computePostpone(
                        workflow, Duration.ofSeconds(30), Duration.ofSeconds(3600));

        assertEquals(
                "When response window is already elapsed, postpone should be 0",
                0L,
                result.getSeconds());
    }

    /**
     * Boundary: taskDef.responseTimeout=0 but taskModel.responseTimeout is non-zero — model wins.
     */
    @Test
    public void computePostponeUsesTaskModelResponseTimeoutWhenTaskDefIsZero() {
        TaskModel task = newTask(TaskType.TASK_TYPE_SIMPLE, TaskModel.Status.IN_PROGRESS);
        // taskDef has responseTimeoutSeconds=0 (not set), taskModel has 120s
        TaskDef taskDef = new TaskDef();
        taskDef.setResponseTimeoutSeconds(0); // explicitly zero
        WorkflowTask workflowTask = new WorkflowTask();
        workflowTask.setTaskDefinition(taskDef);
        task.setWorkflowTask(workflowTask);
        task.setResponseTimeoutSeconds(120);
        task.setStartTime(System.currentTimeMillis());

        WorkflowModel workflow = newWorkflow(Arrays.asList(task), 0);
        Duration result =
                ExecutorUtils.computePostpone(
                        workflow, Duration.ofSeconds(30), Duration.ofSeconds(3600));

        // Should use task model's 120s, not fall back to workflow offset (30s)
        assertTrue(
                "When taskDef.responseTimeout=0, taskModel.responseTimeout (120s) should be used; "
                        + "result="
                        + result.getSeconds(),
                result.getSeconds() > 30);
        assertTrue(
                "Result should be ~120s remaining; got " + result.getSeconds(),
                result.getSeconds() <= 121);
    }

    private WorkflowModel newWorkflow(List<TaskModel> tasks, long timeoutSeconds) {
        WorkflowDef workflowDef = new WorkflowDef();
        workflowDef.setTimeoutSeconds(timeoutSeconds);
        WorkflowModel workflow = new WorkflowModel();
        workflow.setWorkflowId("workflowId");
        workflow.setWorkflowDefinition(workflowDef);
        workflow.setTasks(tasks);
        return workflow;
    }

    @Test
    public void hasInProgressHumanTaskTrueForInProgressHuman() {
        WorkflowModel workflow =
                newWorkflow(
                        Arrays.asList(
                                newTask(TaskType.TASK_TYPE_SIMPLE, TaskModel.Status.COMPLETED),
                                newTask(TaskType.TASK_TYPE_HUMAN, TaskModel.Status.IN_PROGRESS)),
                        0);
        assertTrue(ExecutorUtils.hasInProgressHumanTask(workflow));
    }

    @Test
    public void hasInProgressHumanTaskFalseWhenNoInProgressHuman() {
        WorkflowModel noHuman =
                newWorkflow(
                        Arrays.asList(
                                newTask(TaskType.TASK_TYPE_SIMPLE, TaskModel.Status.IN_PROGRESS)),
                        0);
        assertFalse(ExecutorUtils.hasInProgressHumanTask(noHuman));

        WorkflowModel terminalHuman =
                newWorkflow(
                        Arrays.asList(
                                newTask(TaskType.TASK_TYPE_HUMAN, TaskModel.Status.COMPLETED)),
                        0);
        assertFalse(ExecutorUtils.hasInProgressHumanTask(terminalHuman));
    }

    private TaskModel newTask(String taskType, TaskModel.Status status) {
        TaskModel task = new TaskModel();
        task.setTaskType(taskType);
        task.setStatus(status);
        task.setTaskId("taskId-" + taskType + "-" + status);
        task.setScheduledTime(System.currentTimeMillis());
        return task;
    }
}
