package com.atlassian.bitbucket.mesh.git;

import com.atlassian.bitbucket.mesh.config.ProcessConfig;
import com.atlassian.bitbucket.mesh.event.EventPublisher;
import com.atlassian.bitbucket.mesh.execution.ExecutionManager;
import com.atlassian.bitbucket.mesh.git.builder.GitProcessBuilderFactory;
import com.atlassian.bitbucket.mesh.git.builder.GitProcessBuilderSupport;
import com.atlassian.bitbucket.mesh.git.builder.catfile.GitCatFile;
import com.atlassian.bitbucket.mesh.git.builder.catfile.GitCatFileBatchBuilder;
import com.atlassian.bitbucket.mesh.git.builder.diff.GitDiffTreeBuilder;
import com.atlassian.bitbucket.mesh.git.builder.revlist.FragmentRevListStdioHandler;
import com.atlassian.bitbucket.mesh.git.builder.revlist.GitRevListBuilder;
import com.atlassian.bitbucket.mesh.git.exception.FailedPreconditionException;
import com.atlassian.bitbucket.mesh.git.exception.InvalidArgumentException;
import com.atlassian.bitbucket.mesh.git.exception.UnsupportedCommandException;
import com.atlassian.bitbucket.mesh.repository.ManagedRepository;
import com.atlassian.bitbucket.mesh.repository.MaybeRefsUpdatedEvent;
import com.atlassian.bitbucket.mesh.repository.Repository;
import com.atlassian.bitbucket.mesh.repository.RepositoryManager;
import com.atlassian.bitbucket.mesh.rpc.v1.*;
import com.google.protobuf.ByteString;
import io.grpc.stub.StreamObserver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import java.util.List;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Stream;

import static com.atlassian.bitbucket.mesh.git.GitConstants.COMMAND_NO_REF_UPDATES;
import static com.atlassian.bitbucket.mesh.git.GrpcPorcelainService.computeTotalLength;
import static com.atlassian.bitbucket.mesh.rpc.util.ByteStringUtils.*;
import static java.util.stream.Collectors.joining;
import static org.apache.commons.lang3.StringUtils.isNotBlank;

public class GrpcPlumbingService extends PlumbingServiceGrpc.PlumbingServiceImplBase {

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

    private final RpcArgumentDecoder argumentDecoder;
    private final GitProcessBuilderFactory builderFactory;
    private final EventPublisher eventPublisher;
    private final ExecutionManager executionManager;
    private final ProcessConfig processConfig;
    private final RepositoryManager repositoryManager;

    public GrpcPlumbingService(RpcArgumentDecoder argumentDecoder, GitProcessBuilderFactory builderFactory,
                               EventPublisher eventPublisher, ExecutionManager executionManager,
                               ProcessConfig processConfig, RepositoryManager repositoryManager) {
        this.argumentDecoder = argumentDecoder;
        this.builderFactory = builderFactory;
        this.eventPublisher = eventPublisher;
        this.executionManager = executionManager;
        this.processConfig = processConfig;
        this.repositoryManager = repositoryManager;
    }

    @Override
    public void call(RpcCallRequest request, StreamObserver<RpcStreamFragment> responseObserver) {
        checkCommandAllowed(request);
        maybeTriggerOnRefUpdatedOnCleanup(request);
        builderFactory.builder(request)
                .arguments(argumentDecoder.decodeByteStrings(request.getArgumentsList()))
                .command(request.getCommand())
                .build(new FragmentProcessStdioHandler<>(responseObserver, RpcStreamFragmentFactory.INSTANCE))
                .start();
    }

    @Override
    public StreamObserver<RpcCallStdinFragment> callStdin(StreamObserver<RpcStreamFragment> responseObserver) {
        return new BidirectionalFragmentRequestObserver<>(responseObserver,
                new AbstractStreamFragmentHandler<RpcCallStdinFragment>() {

                    @Nonnull
                    @Override
                    public GitProcessBuilderSupport<?> createBuilder(@Nonnull RpcCallStdinFragment fragment) {
                        requireCase(fragment.getRequestOneofCase(), RpcCallStdinFragment.RequestOneofCase.REQUEST);

                        RpcCallRequest request = fragment.getRequest();
                        checkCommandAllowed(request);
                        maybeTriggerOnRefUpdatedOnCleanup(request);
                        return builderFactory.builder(request)
                                .arguments(argumentDecoder.decodeByteStrings(request.getArgumentsList()))
                                .command(request.getCommand());
                    }

                    @Nonnull
                    @Override
                    public RpcStreamChunk onFragment(@Nonnull RpcCallStdinFragment fragment) {
                        requireCase(fragment.getRequestOneofCase(), RpcCallStdinFragment.RequestOneofCase.STDIN);

                        return fragment.getStdin();
                    }
                });
    }

    @Override
    public void catFile(RpcCatFileRequest request, StreamObserver<RpcStreamFragment> responseObserver) {
        GitProcessBuilderSupport<?> builder;
        if (request.hasPretty()) {
            builder = builderFactory.builder(request).catFile()
                    .pretty()
                    .object(request.getPretty().getObjectId().toStringUtf8());
        } else if (request.hasType()) {
            RpcCatFileType catFileType = request.getType();
            builder = builderFactory.builder(request).catFile()
                    .type()
                    .allowUnknownType(catFileType.getAllowUnknownType())
                    .object(catFileType.getObjectId().toStringUtf8());
        } else {
            throw new UnsupportedOperationException("Use CatFileBatch or CatFileBatchCheck service method instead");
        }

        builder.build(new FragmentProcessStdioHandler<>(responseObserver, RpcStreamFragmentFactory.INSTANCE))
                .start();
    }

    @Override
    public StreamObserver<RpcCatFileStdinFragment> catFileBatch(StreamObserver<RpcStreamFragment> responseObserver) {
        return doCatFileBatch(responseObserver, GitCatFile::batch);
    }

    @Override
    public StreamObserver<RpcCatFileStdinFragment> catFileBatchCheck(
            StreamObserver<RpcStreamFragment> responseObserver) {
        return doCatFileBatch(responseObserver, GitCatFile::batchCheck);
    }

    @Override
    public void diffTree(RpcDiffTreeRequest request, StreamObserver<RpcStreamFragment> responseObserver) {
        GitDiffTreeBuilder builder = builderFactory.builder(request).diffTree()
                .always(request.getAlways())
                .ancestor(emptyToNull(request.getAncestor()))
                .format(emptyToNull(request.getFormat()))
                .merges(request.getMerges())
                .paths(toStringsLazily(request.getPathsList()))
                .recursive(request.getRecursive())
                .rev(request.getRev().toStringUtf8())
                .root(request.getRoot());
        //Apply diff options, if any
        if (request.hasDiffOptions()) {
            builder.applyOptions(request.getDiffOptions());
        }

        builder.build(new FragmentProcessStdioHandler<>(responseObserver, RpcStreamFragmentFactory.INSTANCE))
                .start();
    }

    @Override
    public StreamObserver<RpcDiffTreeStdinFragment> diffTreeStdin(StreamObserver<RpcStreamFragment> responseObserver) {
        return new BidirectionalFragmentRequestObserver<>(responseObserver,
                new AbstractStreamFragmentHandler<RpcDiffTreeStdinFragment>() {

                    @Nonnull
                    @Override
                    public GitProcessBuilderSupport<?> createBuilder(@Nonnull RpcDiffTreeStdinFragment fragment) {
                        requireCase(fragment.getRequestOneofCase(), RpcDiffTreeStdinFragment.RequestOneofCase.REQUEST);

                        RpcDiffTreeRequest request = fragment.getRequest();

                        GitDiffTreeBuilder builder = builderFactory.builder(request).diffTree()
                                .always(request.getAlways())
                                .format(emptyToNull(request.getFormat()))
                                .merges(request.getMerges())
                                .paths(toStringsLazily(request.getPathsList()))
                                .recursive(request.getRecursive())
                                .root(request.getRoot());
                        //Apply diff options, if any
                        if (request.hasDiffOptions()) {
                            builder.applyOptions(request.getDiffOptions());
                        }

                        return builder.stdin(true);
                    }

                    @Nonnull
                    @Override
                    public RpcStreamChunk onFragment(@Nonnull RpcDiffTreeStdinFragment fragment) {
                        requireCase(fragment.getRequestOneofCase(), RpcDiffTreeStdinFragment.RequestOneofCase.STDIN);

                        return fragment.getStdin();
                    }
                });
    }

    @Override
    public void forEachRef(RpcForEachRefRequest request, StreamObserver<RpcStreamFragment> responseObserver) {
        builderFactory.builder(request).forEachRef()
                .count(request.getCount())
                .format(request.getFormat())
                .patterns(toStringsLazily(request.getPatternsList()))
                .sort(request.getSort())
                .build(new FragmentProcessStdioHandler<>(responseObserver, RpcStreamFragmentFactory.INSTANCE))
                .start();
    }

    @Override
    public void lsTree(RpcLsTreeRequest request, StreamObserver<RpcStreamFragment> responseObserver) {
        builderFactory.builder(request).lsTree()
                .nullTerminated(request.getNullTerminated())
                .paths(toStringsLazily(request.getPathsList()))
                .recursive(request.getRecursive())
                .sizes(request.getSizes())
                .tree(request.getTreeish().toStringUtf8())
                .build(new FragmentProcessStdioHandler<>(responseObserver, RpcStreamFragmentFactory.INSTANCE))
                .start();
    }

    @Override
    public void packObjects(RpcPackObjectsRequest request, StreamObserver<RpcStreamFragment> responseObserver) {
        FragmentProcessInputHandler<RpcStreamFragment> handler = new FragmentProcessInputHandler<>(
                responseObserver, RpcStreamFragmentFactory.INSTANCE, toStrings(request.getRevsList()));
        builderFactory.builder(request).packObjects()
                .all(request.getAll())
                .completionHandler(new GitProcessCompletionHandler())
                .deltaBaseOffset(request.getDeltaBaseOffset())
                .revs(true)
                .thin(request.getThin())
                .build(handler)
                .start();
    }

    @Override
    public StreamObserver<RpcPackObjectsStdinFragment> packObjectsStdin(
            StreamObserver<RpcStreamFragment> responseObserver) {
        return new BidirectionalFragmentRequestObserver<>(responseObserver,
                new AbstractStreamFragmentHandler<RpcPackObjectsStdinFragment>() {

                    @Nonnull
                    @Override
                    public GitProcessBuilderSupport<?> createBuilder(@Nonnull RpcPackObjectsStdinFragment fragment) {
                        requireCase(fragment.getRequestOneofCase(), RpcPackObjectsStdinFragment.RequestOneofCase.REQUEST);

                        RpcPackObjectsRequest request = fragment.getRequest();
                        if (request.getRevsCount() > 0) {
                            throw new FailedPreconditionException(
                                    "Use stdin fragments to provide pack-objects revs, not the revs field!");
                        }

                        return builderFactory.builder(request).packObjects()
                                .all(request.getAll())
                                .completionHandler(new GitProcessCompletionHandler())
                                .deltaBaseOffset(request.getDeltaBaseOffset())
                                .revs(true)
                                .thin(request.getThin());
                    }

                    @Nonnull
                    @Override
                    public RpcStreamChunk onFragment(@Nonnull RpcPackObjectsStdinFragment fragment) {
                        requireCase(fragment.getRequestOneofCase(), RpcPackObjectsStdinFragment.RequestOneofCase.STDIN);

                        return fragment.getStdin();
                    }
                });
    }

    @Override
    public void revList(RpcRevListRequest request, StreamObserver<RpcStreamFragment> responseObserver) {
        GitRevListBuilder builder = builderFactory.builder(request).revList();
        if (request.hasRevListOptions()) {
            RpcRevListOptions options = request.getRevListOptions();
            builder.applyOptions(options);

            if (computeTotalLength(options) > processConfig.getCommandLineSize()) {
                //If the paths+revs are too long to fit on the command line, drop the revs and
                //pass them using --stdin instead
                builder.clearRevs()
                        .stdin(true)
                        .build(new FragmentRevListStdioHandler(responseObserver, toStrings(options.getRevsList())))
                        .start();
                return;
            }
        }

        //If we make it here, either no options were supplied or all the files+revs fit on the
        //command line. Either way, we can run the process without using --stdin
        builder.build(new FragmentProcessStdioHandler<>(responseObserver, RpcStreamFragmentFactory.INSTANCE))
                .start();
    }

    @Override
    public StreamObserver<RpcRevListStdinFragment> revListStdin(StreamObserver<RpcStreamFragment> responseObserver) {
        return new BidirectionalFragmentRequestObserver<>(responseObserver,
                new AbstractStreamFragmentHandler<RpcRevListStdinFragment>() {

                    @Nonnull
                    @Override
                    public GitProcessBuilderSupport<?> createBuilder(@Nonnull RpcRevListStdinFragment fragment) {
                        requireCase(fragment.getRequestOneofCase(), RpcRevListStdinFragment.RequestOneofCase.REQUEST);

                        RpcRevListRequest request = fragment.getRequest();

                        GitRevListBuilder builder = builderFactory.builder(request).revList();
                        if (request.hasRevListOptions()) {
                            builder.applyOptions(request.getRevListOptions());
                        }

                        return builder.stdin(true);
                    }

                    @Nonnull
                    @Override
                    public RpcStreamChunk onFragment(@Nonnull RpcRevListStdinFragment fragment) {
                        requireCase(fragment.getRequestOneofCase(), RpcRevListStdinFragment.RequestOneofCase.STDIN);

                        return fragment.getStdin();
                    }
                });
    }

    private static void checkCommandAllowed(RpcCallRequest request) {
        String command = request.getCommand();
        boolean managed = ManagedRepository.Id.matches(request.getRepository());
        Set<String> blockList = managed ? GitConstants.COMMAND_BLOCKED_MANAGED : GitConstants.COMMAND_BLOCKED;

        if (blockList.contains(command)) {
            throw new UnsupportedCommandException(
                    Stream.concat(Stream.of("git", command), toStringsLazily(request.getArgumentsList()).stream())
                            .collect(joining(" ")));
        }

        if (!managed && GitConstants.COMMAND_BLOCKED_MANAGED.contains(command)) {
            //When running as a sidecar, fewer commands are blocked to allow gradual migration of 3rd party plugins.
            //We will however, log a warning here to communicate that the command will not be supported for remote
            //managed repositories.
            log.warn("'git {}' is deprecated is not supported on remote repositories. " +
                    "Update the feature to use the higher level APIs if possible.", command);
        }
    }

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

    private StreamObserver<RpcCatFileStdinFragment> doCatFileBatch(StreamObserver<RpcStreamFragment> responseObserver,
                                                                   Function<GitCatFile, GitCatFileBatchBuilder> catBatchOp) {
        return new BidirectionalFragmentRequestObserver<>(responseObserver,
                new AbstractStreamFragmentHandler<RpcCatFileStdinFragment>() {

                    @Nonnull
                    @Override
                    public GitProcessBuilderSupport<?> createBuilder(@Nonnull RpcCatFileStdinFragment fragment) {
                        requireCase(fragment.getRequestOneofCase(), RpcCatFileStdinFragment.RequestOneofCase.REQUEST);

                        RpcCatFileBatchRequest request = fragment.getRequest();

                        return catBatchOp.apply(builderFactory.builder(request).catFile());
                    }

                    @Nonnull
                    @Override
                    public RpcStreamChunk onFragment(@Nonnull RpcCatFileStdinFragment fragment) {
                        requireCase(fragment.getRequestOneofCase(), RpcCatFileStdinFragment.RequestOneofCase.STDIN);

                        return fragment.getStdin();
                    }
                });
    }

    private void maybeTriggerOnRefUpdatedOnCleanup(RpcCallRequest request) {
        String command = request.getCommand();
        if (command.startsWith("git-")) {
            command = command.substring(4);
        } else if ("git".equals(command)) {
            List<ByteString> arguments = request.getArgumentsList();
            if (arguments.isEmpty()) {
                command = null;
            } else {
                command = arguments.get(0).toStringUtf8();
            }
        }

        boolean readOnly = command != null && COMMAND_NO_REF_UPDATES.contains(command);
        String repositoryId = request.getRepository();
        if (!readOnly && isNotBlank(repositoryId)) {
            //An operation is being executed that may update one or more refs in the target ref. Trigger
            //Repository.onRefsUpdated when the operation finished to trigger repository maintenance tasks such as
            //garbage collection, vacuuming, etc
            executionManager.getContext().ifPresent(context -> context.addCleanupCallback(() -> {
                Repository repository = repositoryManager.findById(repositoryId);
                if (repository != null) {
                    eventPublisher.publish(new MaybeRefsUpdatedEvent(repository));
                }
            }));
        }
    }
}
