package com.atlassian.bitbucket.mesh.client;

import com.atlassian.bitbucket.mesh.git.GitProcessCompletionHandler;
import com.atlassian.bitbucket.mesh.git.builder.GitProcessBuilderFactory;
import com.atlassian.bitbucket.mesh.git.builder.indexpack.IndexPackStdioHandler;
import com.atlassian.bitbucket.mesh.git.exception.AbortedException;
import com.atlassian.bitbucket.mesh.git.exception.InternalFailureException;
import com.atlassian.bitbucket.mesh.gprc.GrpcSupport;
import com.atlassian.bitbucket.mesh.grpc.AbstractFutureResponseObserver;
import com.atlassian.bitbucket.mesh.network.MeshNode;
import com.atlassian.bitbucket.mesh.process.ProcessSummary;
import com.atlassian.bitbucket.mesh.repository.Repository;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcPackObjectsRequest;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcStreamChunk;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcStreamFragment;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcStreamResult;
import com.google.common.base.Throwables;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.charset.CharsetDecoder;
import java.nio.file.Path;
import java.time.Duration;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import static com.atlassian.bitbucket.mesh.git.GitConstants.PATH_OBJECTS;
import static java.nio.charset.StandardCharsets.UTF_8;

/**
 * {@link io.grpc.stub.StreamObserver} used to process the response of {@code PlumbingService/PackObjects}.
 *
 * @see GrpcMeshClient#fetchObjects(MeshNode, Repository, FetchObjectsParameters)
 */
public class PackObjectsResponseObserver
        extends AbstractFutureResponseObserver<RpcPackObjectsRequest, RpcStreamFragment, Path> {

    private static final Logger log = LoggerFactory.getLogger(PackObjectsResponseObserver.class);

    private final GitProcessBuilderFactory builderFactory;
    private final MeshNode node;
    private final FetchObjectsParameters parameters;
    private final Repository repository;
    private final CharBuffer stderr;
    private final CharsetDecoder stderrDecoder;

    private volatile String commandLine;
    private volatile int exitCode;
    private volatile Future<Path> indexPackFuture;
    private volatile IndexPackStdioHandler indexPackHandler;

    public PackObjectsResponseObserver(GitProcessBuilderFactory builderFactory, MeshNode node, Repository repository,
                                       FetchObjectsParameters parameters) {
        this.builderFactory = builderFactory;
        this.node = node;
        this.parameters = parameters;
        this.repository = repository;

        exitCode = -1;
        stderr = CharBuffer.allocate(256);
        stderrDecoder = UTF_8.newDecoder();
    }

    @Override
    public Path asResult() {
        RuntimeException grpcException = null;
        //Wait for the RPC to complete
        try {
            super.asResult();
        } catch (RuntimeException e) {
            if (indexPackFuture == null) {
                //RPC failed before index-pack even started. No need to check whether index-pack finished
                throw e;
            }
            grpcException = e;
        }

        //Remote pack-objects did not throw, but may have returned a non-zero exit code. Check whether the remote
        //process completed successfully.
        stderr.flip();
        new GitProcessCompletionHandler().onComplete(new ProcessSummary.Builder(commandLine)
                .exitCode(exitCode)
                .stderr(stderr.toString().trim())
                .build());

        //Wait for index-pack to complete as well
        RuntimeException indexPackException;
        try {
            if (indexPackFuture != null) {
                return indexPackFuture.get();
            }
            indexPackException = new InternalFailureException("git index-pack was never started");
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            indexPackException = new AbortedException("Interrupted while waiting for index-pack to complete");
        } catch (ExecutionException e) {
            Throwable cause = e.getCause();
            Throwables.throwIfUnchecked(cause);
            indexPackException = new InternalFailureException("git index-pack failed", cause);
        }

        //If we get here, indexPackException must be non-null
        if (grpcException != null) {
            grpcException.addSuppressed(indexPackException);
            throw grpcException;
        }
        throw indexPackException;
    }

    @Override
    public void onCompleted() {
        //Just in case the server never sends a stdout chunk with a closed flag, close stdin.
        indexPackHandler.closeStdin();
        super.onCompleted();
    }

    @Override
    public void onError(Throwable t) {
        super.onError(t);

        if (indexPackFuture != null && !indexPackFuture.isDone()) {
            //Cancel index-pack if the RPC finishes with an error
            indexPackFuture.cancel(false);
        }
    }

    @Override
    public void onNext(RpcStreamFragment fragment) {
        switch (fragment.getFragmentOneofCase()) {
            case START:
                //The server has started git pack-objects
                commandLine = fragment.getStart().getCommandLine().toStringUtf8();
                startIndexPack();
                break;
            case RESULT:
                onResult(fragment.getResult());
                break;
            case STDERR:
                onStderr(fragment.getStderr());
                break;
            case STDOUT:
                onStdout(fragment.getStdout());
                break;
        }
    }

    private void onResult(RpcStreamResult result) {
        log.debug("[{}] pack-objects on {} has finished (exit code {})", repository, node, result.getExitCode());

        exitCode = result.getExitCode();
    }

    private void onStderr(RpcStreamChunk chunk) {
        ByteBuffer last = null;
        for (ByteBuffer buffer : chunk.getData().asReadOnlyByteBufferList()) {
            stderrDecoder.decode(buffer, stderr, false);
            last = buffer;
        }
        if (chunk.getClosed() && last != null) {
            stderrDecoder.decode(last, stderr, true);
        }
    }

    private void onStdout(RpcStreamChunk chunk) {
        indexPackHandler.writeStdin(chunk.getData(), chunk.getClosed());
    }

    private void startIndexPack() {
        if (indexPackHandler == null) {
            indexPackHandler = new IndexPackStdioHandler(repository.getPath().resolve(PATH_OBJECTS));
            indexPackFuture = parameters.applyTimeouts(builderFactory.builder(repository).indexPack()
                    .fixThin(true)
                    .keep(parameters.isKeep())
                    .stdin(true)
                    //default timeouts; these will be overridden by parameters.applyTimeouts if idle/execution
                    //timeout has been set on FetchObjectsParameters
                    .withExecutionInterval(GrpcSupport.maybeReduceToDeadline(Duration.ofHours(24)))
                    .withIdleInterval(Duration.ofHours(10))) // index-pack can go quiet for a long time
                    .build(indexPackHandler)
                    .start();
        } else {
            log.warn("[{}] git index-pack has already been started", repository);
        }
    }
}
