/*
 * Copyright 2020 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.rest.controllers;

import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.UUID;
import java.util.stream.Collectors;

import org.apache.commons.lang3.StringUtils;
import org.conductoross.conductor.model.SignalResponse;
import org.conductoross.conductor.model.WorkflowSignalReturnStrategy;
import org.conductoross.conductor.model.WorkflowStatus;
import org.springframework.http.HttpStatus;
import org.springframework.web.bind.annotation.DeleteMapping;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.PutMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.bind.annotation.RestController;

import com.netflix.conductor.common.metadata.tasks.Task;
import com.netflix.conductor.common.metadata.tasks.TaskType;
import com.netflix.conductor.common.metadata.workflow.RerunWorkflowRequest;
import com.netflix.conductor.common.metadata.workflow.SkipTaskRequest;
import com.netflix.conductor.common.metadata.workflow.StartWorkflowRequest;
import com.netflix.conductor.common.run.*;
import com.netflix.conductor.core.execution.NotificationResult;
import com.netflix.conductor.model.TaskModel;
import com.netflix.conductor.model.WorkflowModel;
import com.netflix.conductor.service.WorkflowService;
import com.netflix.conductor.service.WorkflowTestService;

import io.swagger.v3.oas.annotations.Operation;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import static com.netflix.conductor.rest.config.RequestMappingConstants.WORKFLOW;

import static org.springframework.http.MediaType.APPLICATION_JSON_VALUE;
import static org.springframework.http.MediaType.TEXT_PLAIN_VALUE;

@RestController
@RequestMapping(WORKFLOW)
@Slf4j
public class WorkflowResource {

    private final WorkflowService workflowService;

    private final WorkflowTestService workflowTestService;

    public WorkflowResource(
            WorkflowService workflowService, WorkflowTestService workflowTestService) {
        this.workflowService = workflowService;
        this.workflowTestService = workflowTestService;
    }

    @PostMapping(produces = TEXT_PLAIN_VALUE)
    @Operation(
            summary =
                    "Start a new workflow with StartWorkflowRequest, which allows task to be executed in a domain")
    public String startWorkflow(@RequestBody StartWorkflowRequest request) {
        return workflowService.startWorkflow(request);
    }

    @PostMapping(value = "/{name}", produces = TEXT_PLAIN_VALUE)
    @Operation(
            summary =
                    "Start a new workflow. Returns the ID of the workflow instance that can be later used for tracking")
    public String startWorkflow(
            @PathVariable("name") String name,
            @RequestParam(value = "version", required = false) Integer version,
            @RequestParam(value = "correlationId", required = false) String correlationId,
            @RequestParam(value = "priority", defaultValue = "0", required = false) int priority,
            @RequestBody Map<String, Object> input) {
        return workflowService.startWorkflow(name, version, correlationId, priority, input);
    }

    @SneakyThrows
    @PostMapping(value = "execute/{name}/{version}", produces = APPLICATION_JSON_VALUE)
    @Operation(summary = "Execute a workflow synchronously")
    public Mono<SignalResponse> executeWorkflow(
            @PathVariable("name") String name,
            @PathVariable(value = "version", required = false) Integer version,
            @RequestParam(value = "requestId", required = false) String requestId,
            @RequestParam(value = "waitUntilTaskRef", required = false) String waitUntilTaskRef,
            @RequestParam(value = "waitForSeconds", required = false, defaultValue = "10")
                    Integer waitForSeconds,
            @RequestParam(value = "consistency", required = false, defaultValue = "DURABLE")
                    String workflowConsistency,
            @RequestParam(
                            value = "returnStrategy",
                            required = false,
                            defaultValue = "TARGET_WORKFLOW")
                    WorkflowSignalReturnStrategy returnStrategy,
            @RequestBody StartWorkflowRequest request) {

        if (version == 0) {
            version = null;
        }

        waitForSeconds = Optional.ofNullable(waitForSeconds).filter(w -> w != 0).orElse(10);

        if (StringUtils.isBlank(requestId)) {
            requestId = UUID.randomUUID().toString();
        }

        if (request.getWorkflowDef() != null
                && request.getWorkflowDef().getTasks() != null
                && !request.getWorkflowDef().getTasks().isEmpty()) {
            markAsDynamicWorkflow(request.getInput());
        }

        request.setName(name);
        request.setVersion(version);

        String workflowId = workflowService.startWorkflow(request);
        String workflowRequestId = requestId;

        // Parse comma-separated task refs
        String[] taskRefs =
                StringUtils.isNotBlank(waitUntilTaskRef)
                        ? waitUntilTaskRef.split(",")
                        : new String[0];

        // Poll every 100ms using Flux.interval
        return Flux.interval(Duration.ofMillis(100))
                .map(tick -> workflowService.getWorkflowModel(workflowId, true))
                .filter(
                        workflow -> {
                            // Check if workflow is terminal
                            if (workflow.getStatus().isTerminal()) {
                                return true;
                            }

                            // Check recursively for blocking tasks in the workflow and all
                            // sub-workflows
                            BlockingTaskResult blockingResult =
                                    findBlockingTasks(workflow, taskRefs);
                            return blockingResult.hasBlockingTasks();
                        })
                .next() // Take the first matching workflow
                .map(
                        workflow -> {
                            // Find blocking tasks and blocking workflow recursively
                            BlockingTaskResult blockingResult =
                                    findBlockingTasks(workflow, taskRefs);

                            // Create NotificationResult and return response based on strategy
                            NotificationResult result =
                                    NotificationResult.builder()
                                            .targetWorkflow(workflow)
                                            .blockingWorkflow(
                                                    blockingResult.blockingWorkflow != null
                                                            ? blockingResult.blockingWorkflow
                                                            : workflow)
                                            .blockingTasks(blockingResult.blockingTasks)
                                            .build();

                            return result.toResponse(returnStrategy, workflowRequestId);
                        })
                .timeout(
                        Duration.ofSeconds(waitForSeconds),
                        Mono.defer(
                                () -> {
                                    log.info("Execution timed out for {}", workflowId);
                                    // Timeout reached, return current state
                                    var workflow =
                                            workflowService.getWorkflowModel(workflowId, true);
                                    NotificationResult result =
                                            NotificationResult.builder()
                                                    .targetWorkflow(workflow)
                                                    .blockingWorkflow(workflow)
                                                    .blockingTasks(new ArrayList<>())
                                                    .build();
                                    return Mono.just(
                                            result.toResponse(returnStrategy, workflowRequestId));
                                }));
    }

    @GetMapping("/{name}/correlated/{correlationId}")
    @Operation(summary = "Lists workflows for the given correlation id")
    public List<Workflow> getWorkflows(
            @PathVariable("name") String name,
            @PathVariable("correlationId") String correlationId,
            @RequestParam(value = "includeClosed", defaultValue = "false", required = false)
                    boolean includeClosed,
            @RequestParam(value = "includeTasks", defaultValue = "false", required = false)
                    boolean includeTasks) {
        return workflowService.getWorkflows(name, correlationId, includeClosed, includeTasks);
    }

    @GetMapping("/{workflowId}/tasks")
    @Operation(summary = "Gets the workflow tasks by workflow (execution) id")
    public SearchResult<Task> getExecutionStatusTaskList(
            @PathVariable("workflowId") String workflowId,
            final @RequestParam(value = "start", defaultValue = "0", required = false) Integer
                            start,
            final @RequestParam(value = "count", defaultValue = "15", required = false) Integer
                            count,
            final @RequestParam(value = "status", required = false) List<String> status) {
        Workflow workflow = workflowService.getExecutionStatus(workflowId, true);

        List<Task> workflowFilteredTasks = workflow.getTasks();
        if (status != null && !status.isEmpty()) {
            workflowFilteredTasks =
                    workflow.getTasks().stream()
                            .filter(
                                    t ->
                                            status.stream()
                                                    .map(String::toUpperCase)
                                                    .anyMatch(s -> t.getStatus().name().equals(s)))
                            .collect(Collectors.toList());
        }

        int totalHits = workflowFilteredTasks.size();
        int fromIndex = Math.min(start, totalHits);
        int toIndex = Math.min(start + count, totalHits);
        List<Task> requestedSubList = workflowFilteredTasks.subList(fromIndex, toIndex);
        return new SearchResult<>(totalHits, requestedSubList);
    }

    @PostMapping(value = "/{name}/correlated")
    @Operation(summary = "Lists workflows for the given correlation id list")
    public Map<String, List<Workflow>> getWorkflows(
            @PathVariable("name") String name,
            @RequestParam(value = "includeClosed", defaultValue = "false", required = false)
                    boolean includeClosed,
            @RequestParam(value = "includeTasks", defaultValue = "false", required = false)
                    boolean includeTasks,
            @RequestBody List<String> correlationIds) {
        return workflowService.getWorkflows(name, includeClosed, includeTasks, correlationIds);
    }

    @GetMapping("/{workflowId}")
    @Operation(summary = "Gets the workflow by workflow id")
    public Workflow getExecutionStatus(
            @PathVariable("workflowId") String workflowId,
            @RequestParam(value = "includeTasks", defaultValue = "true", required = false)
                    boolean includeTasks) {
        return workflowService.getExecutionStatus(workflowId, includeTasks);
    }

    @GetMapping("/{workflowId}/status")
    @Operation(summary = "Gets the workflow status summary by workflow (execution) id")
    public WorkflowStatus getWorkflowStatusSummary(
            @PathVariable("workflowId") String workflowId,
            @RequestParam(value = "includeOutput", defaultValue = "false", required = false)
                    boolean includeOutput,
            @RequestParam(value = "includeVariables", defaultValue = "false", required = false)
                    boolean includeVariables) {
        Workflow workflow = workflowService.getExecutionStatus(workflowId, false);
        return new WorkflowStatus(workflow, includeOutput, includeVariables);
    }

    @DeleteMapping("/{workflowId}/remove")
    @Operation(summary = "Removes the workflow from the system")
    public void delete(
            @PathVariable("workflowId") String workflowId,
            @RequestParam(value = "archiveWorkflow", defaultValue = "true", required = false)
                    boolean archiveWorkflow) {
        workflowService.deleteWorkflow(workflowId, archiveWorkflow);
    }

    @GetMapping("/running/{name}")
    @Operation(summary = "Retrieve all the running workflows")
    public List<String> getRunningWorkflow(
            @PathVariable("name") String workflowName,
            @RequestParam(value = "version", defaultValue = "1", required = false) int version,
            @RequestParam(value = "startTime", required = false) Long startTime,
            @RequestParam(value = "endTime", required = false) Long endTime) {
        return workflowService.getRunningWorkflows(workflowName, version, startTime, endTime);
    }

    @PutMapping("/decide/{workflowId}")
    @Operation(summary = "Starts the decision task for a workflow")
    public void decide(@PathVariable("workflowId") String workflowId) {
        workflowService.decideWorkflow(workflowId);
    }

    @PutMapping("/{workflowId}/pause")
    @Operation(summary = "Pauses the workflow")
    public void pauseWorkflow(@PathVariable("workflowId") String workflowId) {
        workflowService.pauseWorkflow(workflowId);
    }

    @PutMapping("/{workflowId}/resume")
    @Operation(summary = "Resumes the workflow")
    public void resumeWorkflow(@PathVariable("workflowId") String workflowId) {
        workflowService.resumeWorkflow(workflowId);
    }

    @PutMapping("/{workflowId}/skiptask/{taskReferenceName}")
    @Operation(summary = "Skips a given task from a current running workflow")
    public void skipTaskFromWorkflow(
            @PathVariable("workflowId") String workflowId,
            @PathVariable("taskReferenceName") String taskReferenceName,
            SkipTaskRequest skipTaskRequest) {
        workflowService.skipTaskFromWorkflow(workflowId, taskReferenceName, skipTaskRequest);
    }

    @PostMapping(value = "/{workflowId}/rerun", produces = TEXT_PLAIN_VALUE)
    @Operation(summary = "Reruns the workflow from a specific task")
    public String rerun(
            @PathVariable("workflowId") String workflowId,
            @RequestBody RerunWorkflowRequest request) {
        return workflowService.rerunWorkflow(workflowId, request);
    }

    @PostMapping("/{workflowId}/restart")
    @Operation(summary = "Restarts a completed workflow")
    @ResponseStatus(
            value = HttpStatus.NO_CONTENT) // for backwards compatibility with 2.x client which
    // expects a 204 for this request
    public void restart(
            @PathVariable("workflowId") String workflowId,
            @RequestParam(value = "useLatestDefinitions", defaultValue = "false", required = false)
                    boolean useLatestDefinitions) {
        workflowService.restartWorkflow(workflowId, useLatestDefinitions);
    }

    @PostMapping("/{workflowId}/retry")
    @Operation(summary = "Retries the last failed task")
    @ResponseStatus(
            value = HttpStatus.NO_CONTENT) // for backwards compatibility with 2.x client which
    // expects a 204 for this request
    public void retry(
            @PathVariable("workflowId") String workflowId,
            @RequestParam(
                            value = "resumeSubworkflowTasks",
                            defaultValue = "false",
                            required = false)
                    boolean resumeSubworkflowTasks) {
        workflowService.retryWorkflow(workflowId, resumeSubworkflowTasks);
    }

    @PostMapping("/{workflowId}/resetcallbacks")
    @Operation(summary = "Resets callback times of all non-terminal SIMPLE tasks to 0")
    @ResponseStatus(
            value = HttpStatus.NO_CONTENT) // for backwards compatibility with 2.x client which
    // expects a 204 for this request
    public void resetWorkflow(@PathVariable("workflowId") String workflowId) {
        workflowService.resetWorkflow(workflowId);
    }

    @DeleteMapping("/{workflowId}")
    @Operation(summary = "Terminate workflow execution")
    public void terminate(
            @PathVariable("workflowId") String workflowId,
            @RequestParam(value = "reason", required = false) String reason) {
        workflowService.terminateWorkflow(workflowId, reason);
    }

    @DeleteMapping("/{workflowId}/terminate-remove")
    @Operation(summary = "Terminate workflow execution and remove the workflow from the system")
    public void terminateRemove(
            @PathVariable("workflowId") String workflowId,
            @RequestParam(value = "reason", required = false) String reason,
            @RequestParam(value = "archiveWorkflow", defaultValue = "true", required = false)
                    boolean archiveWorkflow) {
        workflowService.terminateRemove(workflowId, reason, archiveWorkflow);
    }

    @Operation(
            summary = "Search for workflows based on payload and other parameters",
            description =
                    "use sort options as sort=<field>:ASC|DESC e.g. sort=name&sort=workflowId:DESC."
                            + " If order is not specified, defaults to ASC.")
    @GetMapping(value = "/search")
    public SearchResult<WorkflowSummary> search(
            @RequestParam(value = "start", defaultValue = "0", required = false) int start,
            @RequestParam(value = "size", defaultValue = "100", required = false) int size,
            @RequestParam(value = "sort", required = false) String sort,
            @RequestParam(value = "freeText", defaultValue = "*", required = false) String freeText,
            @RequestParam(value = "query", required = false) String query) {
        return workflowService.searchWorkflows(start, size, sort, freeText, query);
    }

    @Operation(
            summary = "Search for workflows based on payload and other parameters",
            description =
                    "use sort options as sort=<field>:ASC|DESC e.g. sort=name&sort=workflowId:DESC."
                            + " If order is not specified, defaults to ASC.")
    @GetMapping(value = "/search-v2")
    public SearchResult<Workflow> searchV2(
            @RequestParam(value = "start", defaultValue = "0", required = false) int start,
            @RequestParam(value = "size", defaultValue = "100", required = false) int size,
            @RequestParam(value = "sort", required = false) String sort,
            @RequestParam(value = "freeText", defaultValue = "*", required = false) String freeText,
            @RequestParam(value = "query", required = false) String query) {
        return workflowService.searchWorkflowsV2(start, size, sort, freeText, query);
    }

    @Operation(
            summary = "Search for workflows based on task parameters",
            description =
                    "use sort options as sort=<field>:ASC|DESC e.g. sort=name&sort=workflowId:DESC."
                            + " If order is not specified, defaults to ASC")
    @GetMapping(value = "/search-by-tasks")
    public SearchResult<WorkflowSummary> searchWorkflowsByTasks(
            @RequestParam(value = "start", defaultValue = "0", required = false) int start,
            @RequestParam(value = "size", defaultValue = "100", required = false) int size,
            @RequestParam(value = "sort", required = false) String sort,
            @RequestParam(value = "freeText", defaultValue = "*", required = false) String freeText,
            @RequestParam(value = "query", required = false) String query) {
        return workflowService.searchWorkflowsByTasks(start, size, sort, freeText, query);
    }

    @Operation(
            summary = "Search for workflows based on task parameters",
            description =
                    "use sort options as sort=<field>:ASC|DESC e.g. sort=name&sort=workflowId:DESC."
                            + " If order is not specified, defaults to ASC")
    @GetMapping(value = "/search-by-tasks-v2")
    public SearchResult<Workflow> searchWorkflowsByTasksV2(
            @RequestParam(value = "start", defaultValue = "0", required = false) int start,
            @RequestParam(value = "size", defaultValue = "100", required = false) int size,
            @RequestParam(value = "sort", required = false) String sort,
            @RequestParam(value = "freeText", defaultValue = "*", required = false) String freeText,
            @RequestParam(value = "query", required = false) String query) {
        return workflowService.searchWorkflowsByTasksV2(start, size, sort, freeText, query);
    }

    @Operation(
            summary =
                    "Get the uri and path of the external storage where the workflow payload is to be stored")
    @GetMapping({"/externalstoragelocation", "external-storage-location"})
    public ExternalStorageLocation getExternalStorageLocation(
            @RequestParam("path") String path,
            @RequestParam("operation") String operation,
            @RequestParam("payloadType") String payloadType) {
        return workflowService.getExternalStorageLocation(path, operation, payloadType);
    }

    @PostMapping(value = "test", produces = APPLICATION_JSON_VALUE)
    @Operation(summary = "Test workflow execution using mock data")
    public Workflow testWorkflow(@RequestBody WorkflowTestRequest request) {
        return workflowTestService.testWorkflow(request);
    }

    private static void markAsDynamicWorkflow(Map<String, Object> workflowInput) {
        String systemMetadataKey = "_systemMetadata";
        Map<String, Object> systemMetadata;

        if (workflowInput.containsKey(systemMetadataKey)) {
            systemMetadata = (Map<String, Object>) workflowInput.get(systemMetadataKey);
        } else {
            systemMetadata = new HashMap<>();
        }

        systemMetadata.put("dynamic", true);
        workflowInput.put(systemMetadataKey, systemMetadata);
    }

    /**
     * Recursively finds blocking tasks in the workflow and all its sub-workflows. A blocking task
     * is: 1. A WAIT task that is not in terminal state 2. A task matching waitUntilTaskRef that is
     * in terminal state
     *
     * @param workflow The workflow to search
     * @param taskRefs Array of task reference names to wait for
     * @return BlockingTaskResult containing blocking tasks and the workflow where they were found
     */
    private BlockingTaskResult findBlockingTasks(WorkflowModel workflow, String[] taskRefs) {
        List<TaskModel> blockingTasks = new ArrayList<>();
        WorkflowModel blockingWorkflow = null;

        // Check tasks in the current workflow
        for (TaskModel task : workflow.getTasks()) {
            // Check for WAIT tasks that are not terminal (actively waiting)
            if (TaskType.TASK_TYPE_WAIT.equals(task.getTaskType())
                    && !task.getStatus().isTerminal()
                    && task.getWaitTimeout() == 0) {
                blockingTasks.add(task);
                if (blockingWorkflow == null) {
                    blockingWorkflow = workflow;
                }
            }

            // Check if this task matches any of the specified task refs and is terminal
            for (String taskRef : taskRefs) {
                String trimmedRef = taskRef.trim();
                if (trimmedRef.equals(task.getReferenceTaskName())
                        && task.getStatus().isTerminal()) {
                    blockingTasks.add(task);
                    if (blockingWorkflow == null) {
                        blockingWorkflow = workflow;
                    }
                }
            }

            // If this is a SUB_WORKFLOW task, recursively check the sub-workflow
            if (TaskType.TASK_TYPE_SUB_WORKFLOW.equals(task.getTaskType())
                    && StringUtils.isNotBlank(task.getSubWorkflowId())
                    && !task.getStatus().isTerminal()) {
                try {
                    WorkflowModel subWorkflow =
                            workflowService.getWorkflowModel(task.getSubWorkflowId(), true);
                    BlockingTaskResult subResult = findBlockingTasks(subWorkflow, taskRefs);
                    if (subResult.hasBlockingTasks()) {
                        blockingTasks.addAll(subResult.blockingTasks);
                        if (blockingWorkflow == null) {
                            blockingWorkflow = subResult.blockingWorkflow;
                        }
                    }
                } catch (Exception e) {
                    // If we can't fetch the sub-workflow, skip it
                }
            }
        }

        return new BlockingTaskResult(blockingTasks, blockingWorkflow);
    }

    /** Result of finding blocking tasks in a workflow hierarchy */
    private static class BlockingTaskResult {
        final List<TaskModel> blockingTasks;
        final WorkflowModel blockingWorkflow;

        BlockingTaskResult(List<TaskModel> blockingTasks, WorkflowModel blockingWorkflow) {
            this.blockingTasks = blockingTasks;
            this.blockingWorkflow = blockingWorkflow;
        }

        boolean hasBlockingTasks() {
            return blockingTasks != null && !blockingTasks.isEmpty();
        }
    }
}
