package com.atlassian.bitbucket.mesh.process.nu;

import com.atlassian.bitbucket.mesh.execution.ExecutionManager;
import com.atlassian.bitbucket.mesh.process.ProcessConfigurer;
import com.atlassian.bitbucket.mesh.process.ProcessCreator;
import com.atlassian.bitbucket.mesh.process.ProcessParameters;
import com.atlassian.bitbucket.mesh.throttle.ThrottleService;
import com.atlassian.bitbucket.mesh.throttle.ThrottledResource;
import com.atlassian.bitbucket.mesh.throttle.Ticket;
import com.atlassian.bitbucket.mesh.util.ConcurrencyUtils;
import com.atlassian.util.profiling.Ticker;
import com.atlassian.util.profiling.Timers;
import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.MoreExecutors;
import com.zaxxer.nuprocess.NuProcessBuilder;
import com.zaxxer.nuprocess.NuProcessHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import java.util.List;
import java.util.concurrent.*;

import static com.atlassian.bitbucket.mesh.MeshConstants.EXECUTOR_PROCESS_PUMP;
import static com.atlassian.bitbucket.mesh.MeshConstants.EXECUTOR_PROCESS_TIMER;

public class NuProcessCreator implements ProcessCreator {

    private final ExecutionManager executionManager;
    private final ExecutorService asyncExecutor;
    private final List<ProcessConfigurer> configurers;
    private final ScheduledExecutorService timeoutExecutor;
    private final ThrottleService throttleService;

    public NuProcessCreator(ProcessConfigurer[] configurers, ExecutionManager executionManager,
                            ThrottleService throttleService) {
        this.configurers = ImmutableList.copyOf(configurers);
        this.executionManager = executionManager;
        this.throttleService = throttleService;

        asyncExecutor = Executors.newCachedThreadPool(ConcurrencyUtils.createDaemonThreadFactory(EXECUTOR_PROCESS_PUMP));
        timeoutExecutor = ConcurrencyUtils.createScheduledExecutor(EXECUTOR_PROCESS_TIMER);
    }

    @Override
    public <T> T run(@Nonnull ProcessParameters<T> parameters) {
        applyConfigurers(parameters);

        StdioNuProcessHandler<T> handler = new StdioNuProcessHandler<>(parameters, timeoutExecutor);
        NuProcessBuilder builder = createProcessBuilder(parameters, handler);
        try (Ticket ignored1 = maybeAcquireTicket(parameters.getThrottledResource());
             Ticker ignored2 = Timers.start(parameters.toString())) { //Second to not include throttling
            builder.run();
        }

        return handler.asResult();
    }

    /**
     * Terminates the {@link ScheduledExecutorService} which is used to apply execution and idle timeouts.
     */
    public void shutdown() {
        Logger logger = LoggerFactory.getLogger(getClass());
        try {
            if (MoreExecutors.shutdownAndAwaitTermination(asyncExecutor, 10L, TimeUnit.SECONDS)) {
                logger.debug("The async executor has shutdown");
            } else {
                logger.warn("Failed to shutdown the async executor");
            }
        } finally {
            if (MoreExecutors.shutdownAndAwaitTermination(timeoutExecutor, 10L, TimeUnit.SECONDS)) {
                logger.debug("The timeout executor has shutdown");
            } else {
                logger.warn("Failed to shutdown the timeout executor");
            }
        }
    }

    @Nonnull
    @Override
    public <T> Future<T> start(@Nonnull ProcessParameters<T> parameters) {
        applyConfigurers(parameters);

        LatchedStdioNuProcessHandler<T> handler = new LatchedStdioNuProcessHandler<>(parameters, timeoutExecutor);
        NuProcessBuilder builder = createProcessBuilder(parameters, handler);

        //If a ticket is required, acquire it on the request thread instead of the async thread. This ensures
        //the process is never created if the ticket is rejected, so we do not end up with zombie processes
        Ticket ticket = maybeAcquireTicket(parameters.getThrottledResource());

        //Asynchronous processes are pumped using an ExecutorService. NuProcess's pump threads are not
        //used because handlers are simpler to write if they can sporadically block, but blocking isn't
        //safe with the shared thread approach NuProcess uses.
        asyncExecutor.execute(executionManager.maybeWrap(() -> {
            try (Ticket ignored1 = ticket;
                 Ticker ignored2 = Timers.start(parameters.toString())) { //Second to not include throttling
                builder.run();
            }
        }));

        return handler.asFuture();
    }

    private static NuProcessBuilder createProcessBuilder(ProcessParameters<?> parameters, NuProcessHandler handler) {
        //Using this NuProcessBuilder constructor will automatically copy over the environment from the JVM
        //process, and then environment().putAll will override any collisions.
        NuProcessBuilder builder = new NuProcessBuilder(handler, parameters.getArguments());
        builder.environment().putAll(parameters.getEnvironment());
        builder.setCwd(parameters.getWorkDir());

        return builder;
    }

    private void applyConfigurers(ProcessParameters<?> parameters) {
        //Apply any registered configurers. These may modify the environment or working directory, or update
        //timeouts, but they cannot change the arguments, completion handler or I/O handler
        configurers.forEach(configurer -> configurer.configure(parameters));
    }

    private Ticket maybeAcquireTicket(ThrottledResource resource) {
        if (resource == null) {
            return null;
        }
        return throttleService.acquireTicket(resource);
    }
}
