package com.atlassian.bitbucket.mesh.git;

import com.atlassian.bitbucket.mesh.execution.GrpcAsyncExecutor;
import com.atlassian.bitbucket.mesh.git.ref.*;
import com.atlassian.bitbucket.mesh.grpc.BackoffStreamObserver;
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 java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import static com.atlassian.bitbucket.mesh.rpc.util.ByteStringUtils.toByteStringsLazily;
import static com.atlassian.bitbucket.mesh.rpc.util.ByteStringUtils.toStringsLazily;
import static org.apache.commons.lang3.StringUtils.defaultString;

public class GrpcRefService extends RefServiceGrpc.RefServiceImplBase {

    private final GrpcAsyncExecutor executor;
    private final RefManager refManager;
    private final RepositoryManager repositoryManager;

    public GrpcRefService(GrpcAsyncExecutor executor, RefManager refManager, RepositoryManager repositoryManager) {
        this.executor = executor;
        this.repositoryManager = repositoryManager;
        this.refManager = refManager;
    }

    @Override
    public void checkFormat(RpcCheckFormatRequest request, StreamObserver<RpcCheckFormatResponse> responseObserver) {
        refManager.checkFormat(request.getRefId().toStringUtf8());

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

    @Override
    public void createBranch(RpcCreateBranchRequest request, StreamObserver<RpcCreateBranchResponse> responseObserver) {
        Repository repository = repositoryManager.getById(request.getRepository());
        Branch branch = refManager.createBranch(new CreateBranchParameters.Builder(repository, request.getName().toStringUtf8())
                .author(new Person(request.getAuthor()))
                .force(request.getForce())
                .reflog(request.getReflog())
                .startPoint(request.hasStartPoint() ? request.getStartPoint().toStringUtf8() : null)
                .timeouts(request.getTimeouts())
                .build());

        responseObserver.onNext(RpcCreateBranchResponse.newBuilder()
                .setBranch(branch.toBranch())
                .build());
        responseObserver.onCompleted();
    }

    @Override
    public void createTag(RpcCreateTagRequest request, StreamObserver<RpcCreateTagResponse> responseObserver) {
        Repository repository = repositoryManager.getById(request.getRepository());
        Tag tag = refManager.createTag(new CreateTagParameters.Builder(repository, request.getName().toStringUtf8())
                .author(new Person(request.getAuthor()))
                .force(request.getForce())
                .message(request.getMessage().toStringUtf8())
                .startPoint(request.hasStartPoint() ? request.getStartPoint().toStringUtf8() : null)
                .timeouts(request.getTimeouts())
                .type(GitTagType.fromRpc(request.getType()))
                .build());

        responseObserver.onNext(RpcCreateTagResponse.newBuilder()
                .setTag(RpcTag.newBuilder()
                        .setId(ByteString.copyFromUtf8(tag.getId()))
                        .setLatestCommit(tag.getLatestCommit())
                        .setHash(defaultString(tag.getHash())))
                .build());
        responseObserver.onCompleted();
    }

    @Override
    public void deleteBranch(RpcDeleteBranchRequest request, StreamObserver<RpcDeleteBranchResponse> responseObserver) {
        Repository repository = repositoryManager.getById(request.getRepository());
        refManager.deleteBranch(new DeleteBranchParameters.Builder(repository, request.getName().toStringUtf8())
                .author(new Person(request.getAuthor()))
                .force(request.getForce())
                .timeouts(request.getTimeouts())
                .build());

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

    @Override
    public void deleteTag(RpcDeleteTagRequest request, StreamObserver<RpcDeleteTagResponse> responseObserver) {
        Repository repository = repositoryManager.getById(request.getRepository());
        refManager.deleteTag(new DeleteTagParameters.Builder(repository, request.getName().toStringUtf8())
                .author(new Person(request.getAuthor()))
                .timeouts(request.getTimeouts())
                .build());

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

    @Override
    public void resolveAnnotatedTags(RpcResolveAnnotatedTagRequest request,
                                     StreamObserver<RpcResolveAnnotatedTagResponseFragment> responseObserver) {
        Repository repository = repositoryManager.getById(request.getRepository());

        ObserverAnnotatedTagCallback callback =
                new ObserverAnnotatedTagCallback(new BackoffStreamObserver<>(responseObserver));
        StreamAnnotatedTagsParameters parameters = new StreamAnnotatedTagsParameters.Builder(repository)
                .maxMessageLength(request.getMaxMessageLength())
                .tagIds(toStringsLazily(request.getTagIdsList()))
                .timeouts(request.getTimeouts())
                .build();

        executor.execute(responseObserver, () -> {
            refManager.streamAnnotatedTags(parameters, callback);

            callback.onEnd();
        });
    }

    @Override
    public void resolveDefaultBranch(RpcResolveDefaultBranchRequest request,
                                     StreamObserver<RpcResolveDefaultBranchResponse> responseObserver) {
        Repository repository = repositoryManager.getById(request.getRepository());
        Branch defaultBranch = refManager.resolveDefaultBranch(repository);

        responseObserver.onNext(RpcResolveDefaultBranchResponse.newBuilder()
                .setBranch(defaultBranch.toBranch())
                .build());
        responseObserver.onCompleted();
    }

    @Override
    public void resolveRefs(RpcResolveRefsRequest request, StreamObserver<RpcResolveRefsResponse> responseObserver) {
        Repository repository = repositoryManager.getById(request.getRepository());
        Map<String, Ref> refsById = refManager.resolveRefs(new ResolveRefsParameters.Builder(repository)
                .branchIds(toStringsLazily(request.getBranchIdsList()))
                .refIds(toStringsLazily(request.getRefIdsList()))
                .tagIds(toStringsLazily(request.getTagIdsList()))
                .timeouts(request.getTimeouts())
                .build());

        RpcResolveRefsResponse.Builder response = RpcResolveRefsResponse.newBuilder();
        refsById.forEach((id, ref) -> {
            RpcResolvedRef.Builder builder = RpcResolvedRef.newBuilder()
                    .setRefId(ByteString.copyFromUtf8(id));
            if (ref instanceof Branch) {
                builder.setBranch(((Branch) ref).toBranch());
            } else if (ref instanceof Tag) {
                builder.setTag(((Tag) ref).toTag());
            } else {
                return;
            }

            response.addResolvedRefs(builder);
        });

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

    @Override
    public void updateRef(RpcUpdateRefRequest request, StreamObserver<RpcUpdateRefResponse> responseObserver) {
        Repository repository = repositoryManager.getById(request.getRepository());
        ProposedRefUpdate update = ProposedRefUpdate.from(request.getRefUpdate());

        refManager.updateRef(new UpdateRefParameters.Builder(repository, update)
                .timeouts(request.getTimeouts())
                .build());

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

    @Override
    public void updateRefs(RpcUpdateRefsRequest request, StreamObserver<RpcUpdateRefsResponse> responseObserver) {
        Repository repository = repositoryManager.getById(request.getRepository());

        UpdateRefsParameters.Builder builder = new UpdateRefsParameters.Builder(repository)
                .timeouts(request.getTimeouts())
                .transactional(request.getTransactional());

        List<String> failedRefIds = new ArrayList<>();
        for (RpcProposedRefUpdate refUpdate : request.getRefUpdatesList()) {
            try {
                builder.updates(ProposedRefUpdate.from(refUpdate));
            } catch (IllegalArgumentException e) {
                if (request.getTransactional()) {
                    responseObserver.onNext(RpcUpdateRefsResponse.newBuilder()
                            .addAllFailedRefIds(request.getRefUpdatesList().stream()
                                    .map(RpcProposedRefUpdate::getRefId)
                                    .collect(Collectors.toList()))
                            .build());
                    responseObserver.onCompleted();
                    return;
                } else {
                    failedRefIds.add(refUpdate.getRefId().toStringUtf8());
                }
            }
        }
        failedRefIds.addAll(refManager.updateRefs(builder.build()));

        responseObserver.onNext(RpcUpdateRefsResponse.newBuilder()
                .addAllFailedRefIds(toByteStringsLazily(failedRefIds))
                .build());
        responseObserver.onCompleted();
    }
}
