package com.atlassian.bitbucket.mesh.git;

import com.atlassian.bitbucket.mesh.config.GrpcConfig;
import com.atlassian.bitbucket.mesh.execution.GrpcAsyncExecutor;
import com.atlassian.bitbucket.mesh.git.builder.GitProcessBuilderFactory;
import com.atlassian.bitbucket.mesh.git.builder.diff.GitDiffBuilder;
import com.atlassian.bitbucket.mesh.git.builder.diff.GitDiffRenames;
import com.atlassian.bitbucket.mesh.git.builder.revlist.GitRevListOrder;
import com.atlassian.bitbucket.mesh.git.commit.Commit;
import com.atlassian.bitbucket.mesh.git.commit.CommitManager;
import com.atlassian.bitbucket.mesh.git.commit.StreamCommitsParameters;
import com.atlassian.bitbucket.mesh.git.exception.InvalidArgumentException;
import com.atlassian.bitbucket.mesh.git.exception.MergeConflictedException;
import com.atlassian.bitbucket.mesh.git.merge.GitMergeStrategy;
import com.atlassian.bitbucket.mesh.git.merge.MergeManager;
import com.atlassian.bitbucket.mesh.git.merge.MergeParameters;
import com.atlassian.bitbucket.mesh.git.pull.*;
import com.atlassian.bitbucket.mesh.git.ref.Branch;
import com.atlassian.bitbucket.mesh.git.ref.RefUpdate;
import com.atlassian.bitbucket.mesh.grpc.BackoffStreamObserver;
import com.atlassian.bitbucket.mesh.grpc.ConversationalRequestObserver;
import com.atlassian.bitbucket.mesh.grpc.RequestMessageHandler;
import com.atlassian.bitbucket.mesh.repository.Repository;
import com.atlassian.bitbucket.mesh.repository.RepositoryManager;
import com.atlassian.bitbucket.mesh.rpc.v1.*;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcPullRequestRescopeOutcome.Declined;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcPullRequestRescopeOutcome.Merged;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcPullRequestRescopeOutcome.Unknown;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcPullRequestRescopeOutcome.Updated;
import com.atlassian.bitbucket.mesh.util.Person;
import io.grpc.stub.StreamObserver;

import javax.annotation.Nonnull;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import static com.atlassian.bitbucket.mesh.rpc.util.ByteStringUtils.toStringsLazily;
import static com.atlassian.bitbucket.mesh.rpc.v1.RpcMergePullRequestFragment.RequestOneofCase.MERGE_REQUEST;
import static com.atlassian.bitbucket.mesh.rpc.v1.RpcMergePullRequestFragment.RequestOneofCase.UPDATE_REF;
import static java.util.stream.Collectors.toList;
import static java.util.stream.Collectors.toSet;

public class GrpcPullRequestService extends PullRequestServiceGrpc.PullRequestServiceImplBase {

    private final GitProcessBuilderFactory builderFactory;
    private final CommitManager commitManager;
    private final GrpcConfig config;
    private final GrpcAsyncExecutor executor;
    private final MergeManager mergeManager;
    private final PullRequestManager pullRequestManager;
    private final RepositoryManager repositoryManager;

    public GrpcPullRequestService(GitProcessBuilderFactory builderFactory, CommitManager commitManager,
                                  GrpcConfig config, GrpcAsyncExecutor executor, MergeManager mergeManager,
                                  PullRequestManager pullRequestManager, RepositoryManager repositoryManager) {
        this.builderFactory = builderFactory;
        this.commitManager = commitManager;
        this.config = config;
        this.executor = executor;
        this.mergeManager = mergeManager;
        this.pullRequestManager = pullRequestManager;
        this.repositoryManager = repositoryManager;
    }

    @Override
    public void batchRescope(RpcBatchRescopePullRequestRequest request,
                             StreamObserver<RpcBatchRescopePullRequestResponseFragment> responseObserver) {
        BatchRescopePullRequestsParameters.Builder builder = new BatchRescopePullRequestsParameters.Builder()
                .timeouts(request.getTimeouts());
        if (request.hasMaxCommitsPerUpdate()) {
            builder.maxCommitsPerUpdate(request.getMaxCommitsPerUpdate());
        }
        if (request.hasMaxMessageLength()) {
            builder.maxMessageLength(request.getMaxMessageLength());
        }

        //We're likely going to find the same repositories over and over again. Cache the repository-id -> repository
        //resolution for efficiency
        Map<String, Repository> cache = new HashMap<>();
        for (RpcPullRequestRescope rescope : request.getRescopesList()) {
            Repository fromRepo = cache.computeIfAbsent(rescope.getFromRepository(), repositoryManager::getById);
            Repository toRepo = cache.computeIfAbsent(rescope.getToRepository(), repositoryManager::getById);
            builder.rescopes(new PullRequestRescope.Builder(rescope.getRescopeId(), fromRepo, toRepo)
                    .oldHashes(rescope.getOldFromCommit(), rescope.getOldToCommit())
                    .newHashes(rescope.hasNewFromCommit() ? rescope.getNewFromCommit() : null,
                            rescope.hasNewToCommit() ? rescope.getNewToCommit() : null)
                    .build());
        }

        ObserverPullRequestRescopeCallback callback =
                new ObserverPullRequestRescopeCallback(new BackoffStreamObserver<>(responseObserver));

        executor.execute(responseObserver, () -> {
            pullRequestManager.rescope(builder.build(), callback);

            //If we get here, no error was encountered and we can complete the RPC
            callback.complete();
        });
    }

    @Override
    public void batchUpdateRefs(RpcBatchUpdatePullRequestRefsRequest request,
                                StreamObserver<RpcBatchUpdatePullRequestRefsResponse> responseObserver) {
        Set<PullRequest> pullRequests = request.getPullRequestsList().stream()
                .map(this::buildPullRequest)
                .collect(toSet());
        List<PullRequestRefUpdates> updates = pullRequestManager.updateRefs(pullRequests);

        RpcBatchUpdatePullRequestRefsResponse.Builder builder = RpcBatchUpdatePullRequestRefsResponse.newBuilder();
        updates.forEach(update -> builder.addRefUpdates(update.toPullRequestRefUpdates()));

        responseObserver.onNext(builder.build());
        responseObserver.onCompleted();
    }

    @Override
    public void delete(RpcDeletePullRequestRequest request,
                       StreamObserver<RpcDeletePullRequestResponse> responseObserver) {
        PullRequest pullRequest = buildPullRequest(request.getPullRequest());
        List<RefUpdate> refUpdates = pullRequestManager.delete(pullRequest);

        RpcDeletePullRequestResponse.Builder builder = RpcDeletePullRequestResponse.newBuilder();
        refUpdates.forEach(refUpdate -> builder.addRefUpdates(refUpdate.toRefUpdate()));

        responseObserver.onNext(builder.build());
        responseObserver.onCompleted();
    }

    @Override
    public void getChanges(RpcGetPullRequestChangesRequest request,
                           StreamObserver<RpcGetPullRequestChangesResponseFragment> responseObserver) {
        ObserverPullRequestChangeCallback callback =
                new ObserverPullRequestChangeCallback(new BackoffStreamObserver<>(responseObserver));
        StreamPullRequestChangesParameters parameters = buildStreamChangesParameters(request);
        executor.execute(responseObserver, () -> {
            pullRequestManager.streamChanges(parameters, callback);

            //If we make it here, no exceptions were thrown trying to stream the data. Call onEnd on the
            //callback to complete the request (and flush any pending data). This isn't in a finally block
            //because if an exception is thrown above this shouldn't be called
            callback.onEnd();
        });
    }

    @Override
    public void getCommits(RpcGetPullRequestCommitsRequest request,
                           StreamObserver<RpcGetPullRequestCommitsResponseFragment> responseObserver) {
        ObserverPullRequestCommitCallback callback =
                new ObserverPullRequestCommitCallback(new BackoffStreamObserver<>(responseObserver));
        StreamCommitsParameters parameters = buildStreamCommitsParameters(request);
        executor.execute(responseObserver, () -> {
            commitManager.streamCommits(parameters, callback);

            //If we make it here, no exceptions were thrown trying to stream the data. Call onEnd on the
            //callback to complete the request (and flush any pending commits). This isn't in
            //a finally block because if an exception is thrown above this shouldn't be called
            callback.onEnd();
        });
    }

    @Override
    public void getDiff(RpcGetPullRequestDiffRequest request, StreamObserver<RpcStreamFragment> responseObserver) {
        PullRequest pullRequest = buildPullRequest(request.getPullRequest());
        PullRequestRef fromRef = pullRequest.getFromRef();
        PullRequestRef toRef = pullRequest.getToRef();

        GitDiffBuilder builder = builderFactory.builder(toRef.getRepository()).diff()
                .ancestor(pullRequestManager.findMergeBase(pullRequest, request.getAllowUnrelated()))
                .paths(toStringsLazily(request.getPathsList()))
                .rev(fromRef.getLatestCommit());
        if (pullRequest.isCrossRepository()) {
            builder.alternate(fromRef.getRepository());
        }
        if (request.hasDiffOptions()) {
            builder.applyOptions(request.getDiffOptions());
        }
        if (request.hasTimeouts()) {
            builder.withTimeouts(request.getTimeouts());
        }

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

    @Override
    public void getEffectiveDiff(RpcGetPullRequestEffectiveDiffRequest request,
                                 StreamObserver<RpcGetPullRequestEffectiveDiffResponse> responseObserver) {
        GetPullRequestEffectiveDiffParameters parameters = buildGetEffectiveDiffParameters(request);
        PullRequestEffectiveDiff effectiveDiff = pullRequestManager.getEffectiveDiff(parameters);

        RpcPullRequestEffectiveDiff.Builder builder = RpcPullRequestEffectiveDiff.newBuilder()
                .setFromCommit(effectiveDiff.getFromCommit())
                .setToCommit(effectiveDiff.getToCommit());
        effectiveDiff.getRefUpdates().forEach(update -> builder.addRefUpdates(update.toRefUpdate()));

        responseObserver.onNext(RpcGetPullRequestEffectiveDiffResponse.newBuilder()
                .setEffectiveDiff(builder)
                .build());
        responseObserver.onCompleted();
    }

    @Override
    public StreamObserver<RpcMergePullRequestFragment> merge(StreamObserver<RpcMergePullRequestResponseFragment> responseObserver) {
        return new ConversationalRequestObserver<>(responseObserver,
                new RequestMessageHandler<RpcMergePullRequestFragment, RpcMergePullRequestResponseFragment>() {

                    private String mergeCommit;
                    private MergeParameters parameters;

                    @Nonnull
                    @Override
                    public RpcMergePullRequestResponseFragment onMessage(@Nonnull RpcMergePullRequestFragment fragment) {
                        if (fragment.getRequestOneofCase() == MERGE_REQUEST) {
                            parameters = buildMergeParameters(fragment.getMergeRequest());

                            try {
                                mergeCommit = mergeManager.merge(parameters);
                            } catch (MergeConflictedException e) {
                                return RpcMergePullRequestResponseFragment.newBuilder()
                                        .setMergeResponse(RpcMergeResponse.newBuilder()
                                                .setConflicts(e.toMergeConflicts())
                                                .build())
                                        .build();
                            }
                            return RpcMergePullRequestResponseFragment.newBuilder()
                                    .setRunMergeHooks(RpcRunPullRequestMergeHooks.newBuilder()
                                            .setMergeCommit(mergeCommit)
                                            .setRepository(parameters.getRepository().getId())
                                            .build())
                                    .build();
                        }
                        if (fragment.getRequestOneofCase() == UPDATE_REF) {
                            Branch updated = mergeManager.updateRef(parameters, mergeCommit);
                            return RpcMergePullRequestResponseFragment.newBuilder()
                                    .setMergeResponse(RpcMergeResponse.newBuilder()
                                            .setBranch(updated.toBranch())
                                            .build())
                                    .build();
                        }
                        throw new InvalidArgumentException("Received [" + fragment.getRequestOneofCase() + "] where [" +
                                MERGE_REQUEST + " or " + UPDATE_REF + "] was expected");
                    }
                });
    }

    @Override
    public void tryMerge(RpcMergePullRequestRequest request,
                         StreamObserver<RpcTryPullRequestMergeResponse> responseObserver) {
        try {
            mergeManager.tryMerge(buildMergeParameters(request));

            responseObserver.onNext(RpcTryPullRequestMergeResponse.getDefaultInstance());
        } catch (MergeConflictedException e) {
            responseObserver.onNext(RpcTryPullRequestMergeResponse.newBuilder()
                    .setConflicts(e.toMergeConflicts())
                    .build());
        }

        responseObserver.onCompleted();
    }

    @Override
    public void updateRefs(RpcUpdatePullRequestRefsRequest request,
                           StreamObserver<RpcUpdatePullRequestRefsResponse> responseObserver) {
        PullRequest pullRequest = buildPullRequest(request.getPullRequest());
        List<RefUpdate> updates = pullRequestManager.updateRefs(pullRequest);

        RpcUpdatePullRequestRefsResponse.Builder builder = RpcUpdatePullRequestRefsResponse.newBuilder();
        updates.forEach(update -> builder.addRefUpdates(update.toRefUpdate()));

        responseObserver.onNext(builder.build());
        responseObserver.onCompleted();
    }

    private GetPullRequestEffectiveDiffParameters buildGetEffectiveDiffParameters(
            RpcGetPullRequestEffectiveDiffRequest request) {
        return new GetPullRequestEffectiveDiffParameters.Builder(buildPullRequest(request.getPullRequest()))
                .allowUnrelated(request.getAllowUnrelated())
                .build();
    }

    private MergeParameters buildMergeParameters(RpcMergePullRequestRequest request) {
        return new MergeParameters.Builder(buildPullRequest(request.getPullRequest()))
                .allowUnrelated(request.getAllowUnrelated())
                .author(new Person(request.getAuthor()))
                .committer(new Person(request.hasCommitter() ? request.getCommitter() : request.getAuthor()))
                .log(request.getLog())
                .maxMessageLength(request.getMaxMessageLength())
                .mergeStrategy(GitMergeStrategy.fromRpc(request.getStrategy()))
                .message(request.getMessage().toStringUtf8())
                .timeouts(request.getTimeouts())
                .build();
    }

    private PullRequestRef buildPullRequestRef(RpcPullRequestRef ref) {
        return new PullRequestRef.Builder(repositoryManager.getById(ref.getRepository()), ref.getId().toStringUtf8())
                .latestCommit(ref.getLatestCommit())
                .build();
    }

    private PullRequest buildPullRequest(RpcPullRequest request) {
        return new PullRequest.Builder(request.getId(), request.getVersion())
                .fromRef(buildPullRequestRef(request.getFromRef()))
                .state(PullRequestState.fromRpc(request.getState()))
                .toRef(buildPullRequestRef(request.getToRef()))
                .build();
    }

    private StreamPullRequestChangesParameters buildStreamChangesParameters(RpcGetPullRequestChangesRequest request) {
        PullRequest pullRequest = buildPullRequest(request.getPullRequest());

        StreamPullRequestChangesParameters.Builder builder = new StreamPullRequestChangesParameters.Builder(pullRequest)
                .allowUnrelated(request.getAllowUnrelated())
                .limit(request.getLimit())
                .paths(toStringsLazily(request.getPathsList()));
        if (request.hasDiffOptions()) {
            RpcDiffOptions options = request.getDiffOptions();

            GitDiffRenames renames = GitDiffRenames.fromRpc(options.getRenames());
            builder.renames(renames);
            if (renames.supportsThreshold() && options.hasRenameThreshold()) {
                builder.renameThreshold(options.getRenameThreshold());
            }
        }

        return builder.build();
    }

    private StreamCommitsParameters buildStreamCommitsParameters(RpcGetPullRequestCommitsRequest request) {
        return new StreamCommitsParameters.Builder(buildPullRequest(request.getPullRequest()))
                .maxMessageLength(request.hasMaxMessageLength() ?
                        request.getMaxMessageLength() :
                        StreamCommitsParameters.UNLIMITED_MESSAGE_LENGTH)
                .limit(request.getLimit())
                .order(GitRevListOrder.fromRpc(request.getOrder()))
                .skip(request.getSkip())
                .timeouts(request.getTimeouts())
                .build();
    }

    private class ObserverPullRequestRescopeCallback implements PullRequestRescopeCallback {

        private final StreamObserver<RpcBatchRescopePullRequestResponseFragment> observer;

        private RpcBatchRescopePullRequestResponseFragment.Builder builder;
        private long fragmentSize;

        ObserverPullRequestRescopeCallback(StreamObserver<RpcBatchRescopePullRequestResponseFragment> observer) {
            this.observer = observer;

            builder = RpcBatchRescopePullRequestResponseFragment.newBuilder();
        }

        @Override
        public void onDeclined(@Nonnull PullRequestRescope rescope) {
            addOutcome(RpcPullRequestRescopeOutcome.newBuilder()
                    .setRescopeId(rescope.getId())
                    .setDeclined(Declined.getDefaultInstance())
                    .build());
        }

        @Override
        public void onMerged(@Nonnull PullRequestRescope rescope, String mergeCommit) {
            addOutcome(RpcPullRequestRescopeOutcome.newBuilder()
                    .setRescopeId(rescope.getId())
                    .setMerged(mergeCommit == null ?
                            Merged.getDefaultInstance() :
                            Merged.newBuilder().setMergeCommit(mergeCommit).build())
                    .build());
        }

        @Override
        public void onUpdated(@Nonnull PullRequestRescope rescope, @Nonnull List<Commit> addedCommits, int totalAdded,
                              @Nonnull List<Commit> removedCommits, int totalRemoved) {
            addOutcome(RpcPullRequestRescopeOutcome.newBuilder()
                    .setRescopeId(rescope.getId())
                    .setUpdated(Updated.newBuilder()
                            .addAllAddedCommits(addedCommits.stream().map(Commit::toCommit).collect(toList()))
                            .addAllRemovedCommits(removedCommits.stream().map(Commit::toCommit).collect(toList()))
                            .setTotalAddedCommits(totalAdded)
                            .setTotalRemovedCommits(totalRemoved))
                    .build());
        }

        @Override
        public void onUnknown(@Nonnull PullRequestRescope rescope) {
            addOutcome(RpcPullRequestRescopeOutcome.newBuilder()
                    .setRescopeId(rescope.getId())
                    .setUnknown(Unknown.getDefaultInstance())
                    .build());
        }

        void complete() {
            if (fragmentSize > 0) {
                sendFragment();
            }
            observer.onCompleted();
        }

        private void addOutcome(RpcPullRequestRescopeOutcome outcome) {
            builder.addOutcomes(outcome);
            fragmentSize += outcome.getSerializedSize();
            maybeSendFragment();
        }

        private void maybeSendFragment() {
            if (fragmentSize > config.getFragmentSize()) {
                sendFragment();
            }
        }

        private void sendFragment() {
            observer.onNext(builder.build());
            builder = RpcBatchRescopePullRequestResponseFragment.newBuilder();
            fragmentSize = 0L;
        }
    }
}
