package com.atlassian.bitbucket.mesh.git;

import com.atlassian.bitbucket.mesh.AbstractStatusException;
import com.atlassian.bitbucket.mesh.git.builder.GitProcessBuilderSupport;
import com.atlassian.bitbucket.mesh.git.exception.InvalidArgumentException;
import com.atlassian.bitbucket.mesh.grpc.StdinObserver;
import com.atlassian.bitbucket.mesh.process.OsProcess;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcStreamChunk;
import io.grpc.Status;
import io.grpc.stub.StreamObserver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import java.nio.ByteBuffer;
import java.util.concurrent.Future;

import static java.util.Objects.requireNonNull;

/**
 * A {@link StreamObserver} implementation which supports <i>bidirectional</i> streaming.
 * <p>
 * For bidirectional streaming, gRPC requires the service to return a {@link StreamObserver} which can receive
 * multiple messages via {@link #onNext}. This observer uses the first fragment to create a builder and start
 * a process, and then it uses subsequent fragments to provide {@code stdin}. {@code stdout} and {@code stderr}
 * for the process are streamed back to the client using a specialized {@link FragmentProcessStdioHandler}.
 * <p>
 * Because gRPC/Protobuf don't allow inheritance or interfaces or other mechanisms to allow for polymorphism in
 * messages, a secondary {@link RpcFragmentHandler} is used to provide an abstraction over how fragments are
 * encoded and allow creating a {@link GitProcessBuilderSupport process builder} from the first fragment and then
 * extracting {@code stdin} {@link RpcStreamChunk chunks} from subsequent fragments.
 *
 * @param <ReqT>  the request fragment type
 * @param <RespT> the response fragment type
 */
class BidirectionalFragmentRequestObserver<ReqT, RespT> implements StreamObserver<ReqT> {

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

    private final RpcFragmentHandler<ReqT, RespT> fragmentHandler;
    private final StreamObserver<RespT> responseObserver;
    private final BidirectionalProcessStdioHandler<RespT> stdioHandler;

    private Future<Void> future;

    BidirectionalFragmentRequestObserver(@Nonnull StreamObserver<RespT> responseObserver,
                                         @Nonnull RpcFragmentHandler<ReqT, RespT> fragmentHandler) {
        this.responseObserver = requireNonNull(responseObserver, "responseObserver");
        this.fragmentHandler = requireNonNull(fragmentHandler, "fragmentHandler");

        //This must be eagerly instantiated because it registers a cancel handler
        stdioHandler = new BidirectionalProcessStdioHandler<>(responseObserver, fragmentHandler);
    }

    @Override
    public void onCompleted() {
        if (future == null) {
            //If the future is null, we never started a process
            log.warn("The call completed without receiving a request to start");
            responseObserver.onCompleted();
        } else {
            //The process may not have fully started, so this may do nothing, but if the process
            //was ever started and stdin wasn't explicitly closed this will ensure it is
            stdioHandler.closeStdin();
        }
        fragmentHandler.onComplete();
    }

    @Override
    public void onError(Throwable t) {
        boolean canceled = Status.fromThrowable(t).getCode() == Status.Code.CANCELLED;
        if (future == null) {
            if (canceled) {
                //Log cancellation quietly; it's not an error for the client to cancel
                log.debug("The interactive client canceled the request before it was sent");
            } else {
                log.warn("The call failed before a request was sent", t);
            }
        } else if (future.isDone()) {
            if (canceled) {
                //Log cancellation quietly; it's not an error for the client to cancel
                log.debug("The interactive client has canceled the request, but the requested process has already completed");
            } else {
                log.warn("The call failed, but the requested process has already completed", t);
            }
        } else {
            if (canceled) {
                //Log cancellation quietly; it's not an error for the client to cancel
                log.debug("The interactive client has canceled the request");
            } else {
                log.error("Canceling the process; the interactive client has failed", t);
            }
            future.cancel(true);
        }
        fragmentHandler.onError(canceled, t);
    }

    @Override
    public void onNext(ReqT fragment) {
        try {
            if (future == null) {
                future = fragmentHandler.createBuilder(fragment)
                        .build(stdioHandler)
                        .start();
            } else {
                RpcStreamChunk stdinChunk = fragmentHandler.onFragment(fragment);
                if (stdinChunk != null) {
                    stdioHandler.writeChunk(stdinChunk);
                }
            }
        } catch (RuntimeException e) {
            if (e instanceof AbstractStatusException) {
                e = ((AbstractStatusException) e).toStatusException();
            }
            responseObserver.onError(e);
        }
    }

    public static <ReqT extends Enum<ReqT>> void requireCase(ReqT actual, ReqT expected) {
        if (actual != expected) {
            throw new InvalidArgumentException("Received [" + actual + "] where [" + expected + "] was expected");
        }
    }

    private static class BidirectionalProcessStdioHandler<RespT> extends FragmentProcessStdioHandler<RespT> {

        private final StdinObserver stdinObserver;

        BidirectionalProcessStdioHandler(@Nonnull StreamObserver<RespT> observer,
                                         @Nonnull RpcFragmentHandler<?, RespT> fragmentHandler) {
            super(observer, fragmentHandler);

            stdinObserver = new StdinObserver();
        }

        @Override
        public void onStart(@Nonnull OsProcess process) {
            super.onStart(process);

            stdinObserver.onStart(process);
        }

        @Override
        public boolean onStdinReady(@Nonnull ByteBuffer buffer) {
            return stdinObserver.onStdinReady(buffer);
        }

        /**
         * Overrides the base {@link FragmentProcessStdioHandler#closeStdin} to potentially defer actually closing
         * {@code stdin} until any {@link #stdinObserver pending data} has been written.
         *
         * @return {@code true} if {@code stdin} has been closed; otherwise, {@code false} if it will be closed later
         */
        @Override
        protected boolean closeStdin() {
            return stdinObserver.closeStdin();
        }

        void writeChunk(@Nonnull RpcStreamChunk chunk) {
            stdinObserver.writeStdin(requireNonNull(chunk, "chunk").getData(), chunk.getClosed());
        }
    }
}
