package com.atlassian.bitbucket.mesh.git;

import com.atlassian.bitbucket.mesh.execution.GrpcAsyncExecutor;
import com.atlassian.bitbucket.mesh.git.builder.revlist.CommitPumpSettings;
import com.atlassian.bitbucket.mesh.git.builder.revlist.GitRevListMerges;
import com.atlassian.bitbucket.mesh.git.builder.revlist.GitRevListOrder;
import com.atlassian.bitbucket.mesh.git.builder.revlist.GitRevListWalk;
import com.atlassian.bitbucket.mesh.git.commit.*;
import com.atlassian.bitbucket.mesh.git.exception.InvalidArgumentException;
import com.atlassian.bitbucket.mesh.git.exception.MergeConflictedException;
import com.atlassian.bitbucket.mesh.git.exception.ObjectNotFoundException;
import com.atlassian.bitbucket.mesh.git.file.EditFileManager;
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.rebase.RebaseManager;
import com.atlassian.bitbucket.mesh.git.rebase.RebaseParameters;
import com.atlassian.bitbucket.mesh.git.ref.Branch;
import com.atlassian.bitbucket.mesh.git.ref.RefManager;
import com.atlassian.bitbucket.mesh.git.revert.RevertManager;
import com.atlassian.bitbucket.mesh.git.revert.RevertParameters;
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.layout.StorageLayout;
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.util.Person;
import com.google.protobuf.ByteString;
import io.grpc.stub.StreamObserver;

import javax.annotation.Nonnull;
import java.time.Instant;
import java.util.Map;
import java.util.Objects;

import static com.atlassian.bitbucket.mesh.rpc.util.ByteStringUtils.emptyToNull;
import static com.atlassian.bitbucket.mesh.rpc.util.ByteStringUtils.toStringsLazily;
import static java.util.stream.Collectors.toList;
import static org.apache.commons.lang3.StringUtils.defaultString;
import static org.apache.commons.lang3.StringUtils.trimToNull;

public class GrpcCommitService extends CommitServiceGrpc.CommitServiceImplBase {

    private final GitAgent agent;
    private final CommitManager commitManager;
    private final EditFileManager editFileManager;
    private final GrpcAsyncExecutor executor;
    private final MergeManager mergeManager;
    private final RefManager refManager;
    private final RebaseManager rebaseManager;
    private final RepositoryManager repositoryManager;
    private final RevertManager revertManager;
    private final StorageLayout storageLayout;

    public GrpcCommitService(GitAgent agent, CommitManager commitManager, EditFileManager editFileManager,
                             GrpcAsyncExecutor executor, MergeManager mergeManager, RebaseManager rebaseManager,
                             RefManager refManager, RepositoryManager repositoryManager, RevertManager revertManager,
                             StorageLayout storageLayout) {
        this.agent = agent;
        this.commitManager = commitManager;
        this.editFileManager = editFileManager;
        this.executor = executor;
        this.mergeManager = mergeManager;
        this.rebaseManager = rebaseManager;
        this.refManager = refManager;
        this.repositoryManager = repositoryManager;
        this.revertManager = revertManager;
        this.storageLayout = storageLayout;
    }

    @Override
    public StreamObserver<RpcEditFileFragment> editFile(StreamObserver<RpcEditFileResponseFragment> responseObserver) {
        return new EditFileRequestObserver(editFileManager, refManager, repositoryManager,
                responseObserver, storageLayout.getTempDir());
    }

    @Override
    public void getCommit(RpcGetCommitRequest request, StreamObserver<RpcGetCommitResponse> responseObserver) {
        Commit commit = commitManager.getCommit(buildGetCommitParameters(request));

        responseObserver.onNext(RpcGetCommitResponse.newBuilder()
                .setCommit(commit.toCommit())
                .build());
        responseObserver.onCompleted();
    }

    @Override
    public void getCommits(RpcGetCommitsRequest request,
                           StreamObserver<RpcGetCommitsResponseFragment> responseObserver) {
        ObserverCommitCallback callback = new ObserverCommitCallback(new BackoffStreamObserver<>(responseObserver));
        StreamCommitsParameters parameters = buildStreamCommitsParameters(request);
        //Execute the server streaming operation on an executor thread so we can receive onReady callbacks (we need to
        //return from this method to get them).
        executor.submit(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 getCommonAncestor(RpcGetCommonAncestorRequest request,
                                  StreamObserver<RpcGetCommonAncestorResponse> responseObserver) {
        String commitId = commitManager.getMergeBase(buildGetMergeBaseParameters(request));

        responseObserver.onNext(RpcGetCommonAncestorResponse.newBuilder()
                .setCommitId(defaultString(commitId))
                .build());
        responseObserver.onCompleted();
    }

    @Override
    public void lastModified(RpcLastModifiedRequest request,
                             StreamObserver<RpcLastModifiedResponseFragment> responseObserver) {
        ObserverLastModifiedCallback callback = new ObserverLastModifiedCallback(
                new BackoffStreamObserver<>(responseObserver));
        StreamLastModifiedParameters parameters = buildStreamLastModifiedParameters(request);
        executor.submit(responseObserver, () -> {
            commitManager.streamLastModified(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 file modifications). This isn't in
            //a finally block because if an exception is thrown above this shouldn't be called
            callback.onEnd();
        });
    }

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

                    private String mergeCommit;
                    private MergeParameters parameters;

                    @Nonnull
                    @Override
                    public RpcMergeResponseFragment onMessage(@Nonnull RpcMergeFragment fragment) {
                        if (fragment.getRequestOneofCase() == RpcMergeFragment.RequestOneofCase.MERGE_REQUEST) {
                            parameters = buildMergeParameters(fragment.getMergeRequest());

                            try {
                                // Perform the merge
                                mergeCommit = mergeManager.merge(parameters);
                            } catch (MergeConflictedException e) {
                                return RpcMergeResponseFragment.newBuilder()
                                        .setMergeResponse(RpcMergeResponse.newBuilder()
                                                .setConflicts(e.toMergeConflicts())
                                                .build())
                                        .build();
                            }
                            return RpcMergeResponseFragment.newBuilder()
                                    .setRunMergeHooks(RpcRunMergeHooks.newBuilder()
                                            //We know these Branch casts are safe because buildMergeParameters
                                            //is our own method and always sets the refs as full Branches
                                            .setFromRef(((Branch) parameters.getFrom()).toBranch())
                                            .setMergeCommit(mergeCommit)
                                            .setRepository(parameters.getRepository().getId())
                                            .setToRef(((Branch) parameters.getTo()).toBranch())
                                            .build())
                                    .build();
                        }
                        if (fragment.getRequestOneofCase() == RpcMergeFragment.RequestOneofCase.UPDATE_REF) {
                            // Update the ref
                            Branch updated = mergeManager.updateRef(parameters, mergeCommit);
                            return RpcMergeResponseFragment.newBuilder()
                                    .setMergeResponse(RpcMergeResponse.newBuilder()
                                            .setBranch(updated.toBranch())
                                            .build())
                                    .build();
                        }
                        throw new InvalidArgumentException("Received [" + fragment.getRequestOneofCase() + "] where [" +
                                RpcMergeFragment.RequestOneofCase.MERGE_REQUEST + " or " +
                                RpcMergeFragment.RequestOneofCase.UPDATE_REF + "] was expected");
                    }
                });
    }

    @Override
    public StreamObserver<RpcRebaseFragment> rebase(StreamObserver<RpcRebaseResponseFragment> responseObserver) {
        return new ConversationalRequestObserver<>(responseObserver,
                new RequestMessageHandler<RpcRebaseFragment, RpcRebaseResponseFragment>() {

                    private String commitId;
                    private RebaseParameters parameters;

                    @Nonnull
                    @Override
                    public RpcRebaseResponseFragment onMessage(@Nonnull RpcRebaseFragment fragment) {
                        if (fragment.getRequestOneofCase() == RpcRebaseFragment.RequestOneofCase.REBASE_REQUEST) {
                            parameters = buildRebaseParameters(fragment.getRebaseRequest());

                            // Perform the rebase
                            commitId = rebaseManager.rebase(parameters);
                            return RpcRebaseResponseFragment.newBuilder()
                                    .setRunRebaseHooks(RpcRunRebaseHooks.newBuilder()
                                            .setBranch(parameters.getBranch().toBranch())
                                            .setCommitId(commitId)
                                            .setRepository(parameters.getRepository().getId()))
                                    .build();
                        }
                        if (fragment.getRequestOneofCase() == RpcRebaseFragment.RequestOneofCase.UPDATE_REF) {
                            // Update the ref
                            Branch updated = rebaseManager.updateRef(parameters, commitId);
                            return RpcRebaseResponseFragment.newBuilder()
                                    .setRebaseResponse(RpcRebaseResponse.newBuilder()
                                            .setBranch(updated.toBranch())
                                            .build())
                                    .build();
                        }
                        throw new InvalidArgumentException("Received [" + fragment.getRequestOneofCase() + "] where [" +
                                RpcRebaseFragment.RequestOneofCase.REBASE_REQUEST + " or " +
                                RpcRebaseFragment.RequestOneofCase.UPDATE_REF + "] was expected");
                    }
                });
    }

    @Override
    public void resolveCommits(RpcResolveCommitsRequest request,
                               StreamObserver<RpcResolveCommitsResponse> responseObserver) {
        Repository repository = repositoryManager.getById(request.getRepository());
        Map<String, String> resolved = commitManager.resolveCommits(new ResolveCommitsParameters.Builder(repository)
                .commitishes(toStringsLazily(request.getCommitishesList()))
                .build());

        RpcResolveCommitsResponse.Builder builder = RpcResolveCommitsResponse.newBuilder();
        resolved.forEach((commitish, commitId) -> builder.addResolvedCommits(
                RpcResolvedCommit.newBuilder().setCommitId(commitId)
                        .setCommitish(ByteString.copyFromUtf8(commitish))
                        .build()));

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

    @Override
    public StreamObserver<RpcRevertFragment> revert(StreamObserver<RpcRevertResponseFragment> responseObserver) {
        return new ConversationalRequestObserver<>(responseObserver,
                new RequestMessageHandler<RpcRevertFragment, RpcRevertResponseFragment>() {

                    private RevertParameters parameters;
                    private String revertCommit;

                    @Nonnull
                    @Override
                    public RpcRevertResponseFragment onMessage(@Nonnull RpcRevertFragment fragment) {
                        if (fragment.getRequestOneofCase() == RpcRevertFragment.RequestOneofCase.REVERT_REQUEST) {
                            parameters = buildRevertParameters(fragment.getRevertRequest());
                            revertCommit = revertManager.revert(parameters);
                            return RpcRevertResponseFragment.newBuilder()
                                    .setRunRevertHooks(RpcRunRevertHooks.newBuilder()
                                            .setCommitId(revertCommit)
                                            .setRepository(parameters.getRepository().getId()))
                                    .build();
                        }
                        if (fragment.getRequestOneofCase() == RpcRevertFragment.RequestOneofCase.UPDATE_REF) {
                            if (revertCommit == null) {
                                throw new InvalidArgumentException("Received [" + fragment.getRequestOneofCase() + "] " +
                                        "without processing of [" +
                                        RpcRevertFragment.RequestOneofCase.REVERT_REQUEST + "]");
                            }
                            Branch branch = revertManager.updateRef(parameters, revertCommit);
                            return RpcRevertResponseFragment.newBuilder()
                                    .setRevertResponse(RpcRevertResponse.newBuilder()
                                            .setBranch(branch.toBranch()))
                                    .build();
                        }
                        throw new InvalidArgumentException("Received [" + fragment.getRequestOneofCase() + "] where [" +
                                RpcRevertFragment.RequestOneofCase.REVERT_REQUEST + " or " +
                                RpcRevertFragment.RequestOneofCase.UPDATE_REF + "] was expected");
                    }
                });
    }

    @Override
    public void traverseCommits(RpcTraverseCommitsRequest request,
                                StreamObserver<RpcTraverseCommitsResponseFragment> responseObserver) {
        ObserverTraversalCallback callback = new ObserverTraversalCallback(
                new BackoffStreamObserver<>(responseObserver));
        TraverseCommitsParameters parameters = buildTraverseParameters(request);
        executor.submit(responseObserver, () -> {
            commitManager.traverseCommits(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 tryMerge(RpcMergeRequest request, StreamObserver<RpcTryMergeResponse> responseObserver) {
        try {
            mergeManager.tryMerge(buildMergeParameters(request));

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

        responseObserver.onCompleted();
    }

    @Override
    public void tryRebase(RpcRebaseRequest request, StreamObserver<RpcTryRebaseResponse> responseObserver) {
        rebaseManager.tryRebase(buildRebaseParameters(request));

        responseObserver.onNext(RpcTryRebaseResponse.getDefaultInstance());
        responseObserver.onCompleted();
    }

    private GetCommitParameters buildGetCommitParameters(RpcGetCommitRequest request) {
        Repository repository = repositoryManager.getById(request.getRepository());

        return new GetCommitParameters.Builder(repository, request.getCommitish().toStringUtf8())
                .maxMessageLength(request.hasMaxMessageLength() ?
                        request.getMaxMessageLength() :
                        CommitPumpSettings.UNLIMITED_MESSAGE_LENGTH)
                .path(request.getPath().toStringUtf8())
                .timeouts(request.getTimeouts())
                .build();
    }

    private GetMergeBaseParameters buildGetMergeBaseParameters(RpcGetCommonAncestorRequest request) {
        Repository repository = repositoryManager.getById(request.getRepository());

        GetMergeBaseParameters.Builder builder = new GetMergeBaseParameters.Builder(repository)
                .commitIds(toStringsLazily(request.getCommitishesList()))
                .required(request.getRequired())
                .timeouts(request.getTimeouts());
        request.getAlternatesList().stream()
                .map(repositoryManager::getById)
                .forEach(builder::alternate);

        return builder.build();
    }

    private MergeParameters buildMergeParameters(RpcMergeRequest request) {
        Repository repository = repositoryManager.getById(request.getRepository());
        Repository fromRepository = request.hasFromRepository() ?
                repositoryManager.getById(request.getFromRepository()) : repository;

        Branch from = resolveRequiredBranch(fromRepository, request.getFromBranch().toStringUtf8());
        Branch to = resolveRequiredBranch(repository, request.getToBranch().toStringUtf8());

        //If the request is to merge a specific revision, replace the latest commit hash on the "from" ref with
        //the explicit revision to merge
        if (trimToNull(request.getFromCommitId()) != null) {
            from = new Branch.Builder(from)
                    .latestCommit(request.getFromCommitId())
                    .build();
        }
        //If the request contains an expected target revision, replace the latest commit hash on the "to" ref with
        //the expected revision
        if (trimToNull(request.getToCommitId()) != null) {
            to = new Branch.Builder(to)
                    .latestCommit(request.getToCommitId())
                    .build();
        }

        return new MergeParameters.Builder(repository)
                .allowUnrelated(request.getAllowUnrelated())
                .author(new Person(request.getAuthor()))
                .committer(new Person(request.hasCommitter() ? request.getCommitter() : request.getAuthor()))
                .from(from)
                .fromRepository(fromRepository)
                .log(request.getLog())
                .maxMessageLength(request.getMaxMessageLength())
                .mergeStrategy(GitMergeStrategy.fromRpc(request.getStrategy()))
                .message(request.getMessage().toStringUtf8())
                .targetRef(defaultString(emptyToNull(request.getTargetRef()), to.getId()))
                .timeouts(request.getTimeouts())
                .to(to)
                .build();
    }

    private RebaseParameters buildRebaseParameters(RpcRebaseRequest request) {
        Repository repository = repositoryManager.getById(request.getRepository());
        Repository upstreamRepository = request.hasUpstreamRepository() ?
                repositoryManager.getById(request.getUpstreamRepository()) : repository;

        Branch branch = resolveRequiredBranch(repository, request.getBranch().toStringUtf8());
        //If the request is to rebase using a specific revision, replace the latest commit hash on the branch with
        //the explicit revision to use
        if (trimToNull(request.getCommitId()) != null) {
            branch = new Branch.Builder(branch)
                    .latestCommit(request.getCommitId())
                    .build();
        }

        return new RebaseParameters.Builder(repository)
                .allowEmpty(request.getAllowEmpty())
                .branch(branch)
                .committer(new Person(request.getCommitter()))
                .timeouts(request.getTimeouts())
                .upstream(request.getUpstream().toStringUtf8())
                .upstreamRepository(upstreamRepository)
                .build();
    }

    private RevertParameters buildRevertParameters(RpcRevertRequest request) {
        Repository repository = repositoryManager.getById(request.getRepository());
        RevertParameters.Builder builder = new RevertParameters.Builder(repository)
                .commitId(request.getCommitId())
                .committer(new Person(request.getCommitter()))
                .timeouts(request.getTimeouts())
                .branch(Branch.fromBranch(request.getBranch()));
        if (request.hasTargetBranch()) {
            builder.targetBranch(request.getTargetBranch());
        }
        if (request.hasMergedCommitParentNumber()) {
            builder.mergedCommitParentNumber(request.getMergedCommitParentNumber());
        }
        return builder.build();
    }

    private StreamCommitsParameters buildStreamCommitsParameters(RpcGetCommitsRequest request) {
        Repository repository = repositoryManager.getById(request.getRepository());

        return new StreamCommitsParameters.Builder(repository)
                .all(request.getAll())
                .alternates(request.getAlternatesList().stream().map(repositoryManager::getById).collect(toList()))
                .branches(request.getBranches())
                .excludes(toStringsLazily(request.getExcludesList()))
                .followRenames(request.getFollowRenames())
                .ignoreMissing(request.getIgnoreMissing())
                .includes(toStringsLazily(request.getIncludesList()))
                .limit(request.getLimit())
                .maxMessageLength(request.hasMaxMessageLength() ?
                        request.getMaxMessageLength() :
                        StreamCommitsParameters.UNLIMITED_MESSAGE_LENGTH)
                .merges(GitRevListMerges.fromRpc(request.getMerges()))
                .order(GitRevListOrder.fromRpc(request.getOrder()))
                .paths(toStringsLazily(request.getPathsList()))
                .since(request.hasSince() ? Instant.ofEpochSecond(request.getSince()) : null)
                .skip(request.getSkip())
                .tags(request.getTags())
                .timeouts(request.getTimeouts())
                .walk(GitRevListWalk.fromRpc(request.getWalk()))
                .build();
    }

    private StreamLastModifiedParameters buildStreamLastModifiedParameters(RpcLastModifiedRequest request) {
        Repository repository = repositoryManager.getById(request.getRepository());

        return new StreamLastModifiedParameters.Builder(repository, request.getCommitish().toStringUtf8())
                .maxMessageLength(request.hasMaxMessageLength() ?
                        request.getMaxMessageLength() :
                        CommitPumpSettings.UNLIMITED_MESSAGE_LENGTH)
                .path(request.getPath().toStringUtf8())
                .timeouts(request.getTimeouts())
                .build();
    }

    private TraverseCommitsParameters buildTraverseParameters(RpcTraverseCommitsRequest request) {
        Repository repository = repositoryManager.getById(request.getRepository());

        boolean ignoreMissing = request.getIgnoreMissing();
        TraverseCommitsParameters.Builder builder = new TraverseCommitsParameters.Builder(repository)
                .all(request.getAll())
                .branches(request.getBranches())
                .exclude(toStringsLazily(request.getExcludesList()))
                .ignoreMissing(ignoreMissing)
                .include(toStringsLazily(request.getIncludesList()))
                .timeouts(request.getTimeouts())
                .tags(request.getTags());

        if (request.getOrder() != RpcRevListOrder.ORDER_UNSPECIFIED) {
            builder.order(GitRevListOrder.fromRpc(request.getOrder()));
        }

        if (ignoreMissing) {
            request.getAlternatesList().stream()
                    .map(repositoryManager::findById)
                    .filter(Objects::nonNull) // skip repositories that could not be found
                    .forEach(builder::alternate);
        } else {
            request.getAlternatesList().stream()
                    .map(repositoryManager::getById)
                    .forEach(builder::alternate);
        }
        return builder.build();
    }

    private Branch resolveRequiredBranch(Repository repository, String branchId) {
        Branch branch = agent.resolveBranch(repository, branchId);
        if (branch == null) {
            throw new ObjectNotFoundException(branchId);
        }

        return branch;
    }
}
