package com.atlassian.bitbucket.mesh.transaction;

import com.atlassian.bitbucket.mesh.git.exception.InvalidArgumentException;
import com.atlassian.bitbucket.mesh.git.hook.GitHookService;
import com.atlassian.bitbucket.mesh.git.hook.HookRequestHandle;
import com.atlassian.bitbucket.mesh.git.pull.PullRequest;
import com.atlassian.bitbucket.mesh.git.pull.PullRequestManager;
import com.atlassian.bitbucket.mesh.git.pull.PullRequestRef;
import com.atlassian.bitbucket.mesh.git.pull.PullRequestState;
import com.atlassian.bitbucket.mesh.git.ref.ProposedRefUpdate;
import com.atlassian.bitbucket.mesh.git.ref.RefManager;
import com.atlassian.bitbucket.mesh.git.ref.UpdateRefsParameters;
import com.atlassian.bitbucket.mesh.git.transcode.TranscodeService;
import com.atlassian.bitbucket.mesh.gprc.GrpcSupport;
import com.atlassian.bitbucket.mesh.repository.*;
import com.atlassian.bitbucket.mesh.rpc.v1.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import java.util.List;
import java.util.stream.Collectors;

import static com.atlassian.bitbucket.mesh.rpc.util.ByteStringUtils.emptyToNull;
import static java.util.Objects.requireNonNull;

public class DefaultRpcTransactionExecutor implements RpcTransactionExecutor {

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

    private final GitHookService hookService;
    private final PullRequestManager pullRequestManager;
    private final RefManager refManager;
    private final RepositoryManager repositoryManager;
    private final TranscodeService transcodeService;

    public DefaultRpcTransactionExecutor(GitHookService hookService,
                                         PullRequestManager pullRequestManager,
                                         RefManager refManager,
                                         RepositoryManager repositoryManager,
                                         TranscodeService transcodeService) {
        this.hookService = hookService;
        this.pullRequestManager = pullRequestManager;
        this.refManager = refManager;
        this.repositoryManager = repositoryManager;
        this.transcodeService = transcodeService;
    }

    @Override
    public void execute(@Nonnull RpcTransaction transaction) {
        requireNonNull(transaction, "transaction");

        switch (transaction.getDetailsOneofCase()) {
            case CREATE_REPOSITORY:
                createRepository(transaction.getCreateRepository());
                break;
            case DELETE_PULL_REQUEST:
                deletePullRequest(transaction.getDeletePullRequest());
                break;
            case DELETE_REPOSITORY:
                deleteRepository(transaction.getDeleteRepository());
                break;
            case SET_DEFAULT_BRANCH:
                setDefaultBranch(transaction.getSetDefaultBranch());
                break;
            case SET_METADATA:
                setMetadata(transaction.getSetMetadata());
                break;
            case SET_TRANSCODE:
                setTranscode(transaction.getSetTranscode());
                break;
            case UPDATE_REFS:
                updateRefs(transaction.getUpdateRefs());
                break;
            default:
                throw new InvalidArgumentException("Received [" + transaction.getDetailsOneofCase() +
                        "] where [CREATE_REPOSITORY, DELETE_PULL_REQUEST, DELETE_REPOSITORY, SET_DEFAULT_BRANCH, " +
                        "SET_METADATA, UPDATE_PULL_REQUEST_REFS or UPDATE_REFS] was expected");
        }
    }

    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 PullRequestRef buildPullRequestRef(RpcPullRequestRef ref) {
        return new PullRequestRef.Builder(repositoryManager.getById(ref.getRepository()), ref.getId().toStringUtf8())
                .latestCommit(ref.getLatestCommit())
                .build();
    }

    private void createRepository(RpcCreateRepositoryRequest request) {
        CreateRepositoryParameters.Builder builder = new CreateRepositoryParameters.Builder(request.getRepository())
                .metadata(request.getMetadataMap());
        if (request.hasDefaultBranch()) {
            builder.defaultBranch(emptyToNull(request.getDefaultBranch()));
        }
        if (request.hasOrigin()) {
            builder.origin(repositoryManager.getById(request.getOrigin()));
        }
        if (request.hasTimeouts()) {
            builder.timeouts(request.getTimeouts());
        } else {
            //Propagate the gRPC deadline as the timeout. We need to explicitly do that here because
            //the timeout is also used as a lock timeout; the default behavior of GitProcessBuilder where the
            //gRPC deadline is used as the default timeout if no timeout is set is not sufficient.
            builder.timeout(GrpcSupport.getTimeout().getSeconds());
        }

        repositoryManager.create(builder.build());
    }

    private void deletePullRequest(RpcDeletePullRequestRequest request) {
        pullRequestManager.delete(buildPullRequest(request.getPullRequest()));
    }

    private void deleteRepository(RpcDeleteRepositoryRequest request) {
        try {
            Repository repository = repositoryManager.getById(request.getRepository());

            DeleteRepositoryParameters.Builder builder = new DeleteRepositoryParameters.Builder(repository);
            if (request.hasHierarchyMembership()) {
                RpcHierarchyMembership membership = request.getHierarchyMembership();
                builder.onlyMember(membership.getOnlyMember());
                if (membership.hasOrigin()) {
                    builder.origin(repositoryManager.getById(membership.getOrigin()));
                    if (membership.hasOnlyForkOfOrigin()) {
                        builder.onlyForkOfOrigin(membership.getOnlyForkOfOrigin());
                    }
                }
                membership.getForksList().forEach(forkId -> builder.forks(repositoryManager.getById(forkId)));
            }

            repositoryManager.delete(builder.build());
        } catch (RepositoryNotFoundException e) {
            log.info("[{}] Repository does not exist. Skipping replay of repository deletion", e.getId());
        }
    }

    private void setDefaultBranch(RpcSetDefaultBranchRequest request) {
        Repository repository = repositoryManager.getById(request.getRepository());
        repositoryManager.setDefaultBranch(repository, request.getBranchId().toStringUtf8());
    }

    private void setMetadata(RpcSetMetadataRequest request) {
        Repository repository = repositoryManager.getById(request.getRepository());
        repositoryManager.setMetadata(repository, request.getMetadataMap());
    }

    private void setTranscode(RpcSetTranscodeEnabledRequest request) {
        Repository repository = repositoryManager.getById(request.getRepository());
        transcodeService.setEnabled(repository, request.getEnabled());
    }

    private void updateRefs(RpcUpdateRefsRequest request) {
        Repository repository = repositoryManager.getById(request.getRepository());
        try (HookRequestHandle ignored = hookService.registerRequest(repository)) {
            List<ProposedRefUpdate> updates = request.getRefUpdatesList().stream()
                    .map(ProposedRefUpdate::from)
                    .collect(Collectors.toList());

            UpdateRefsParameters.Builder builder = new UpdateRefsParameters.Builder(repository)
                    .dereferenceSymbolicRefs(false) //update the refs directly
                    .transactional(true)
                    .updates(updates);

            //A note on timeouts: we propagate the timeouts provided on the RpcUpdateRefsRequest, but we don't actually
            //expect them to be set; when an RpcUpdateRefsRequest is replicated, the timeouts that the leader applied
            //are not passed to the followers (because they're not readily available from the reference-transaction
            //hook that triggers the replication).
            //
            //When no timeouts are provided (the current default), the RPC deadline of the `TransactionService/Transact`
            //call serves as the execution timeout; this is the default behaviour for git processes unless an explicit
            //timeout is set on their builder. We set a long idle timeout here to prevent the coordination overhead that
            //is triggered from the follower's (so this node) reference-transaction hooks.
            //The update-ref process is 'paused' while waiting for the leader's decision to commit or abort. We don't
            //want the default idle timeout to cause the process to be aborted when we're waiting for the leader to
            //decide.
            if (request.hasTimeouts()) {
                builder.timeouts(request.getTimeouts());
            } else {
                //effectively disable the idle timeout; uses the gRPC timeout as the execution timeout
                builder.idleTimeout(36000L); //10 hours
            }

            refManager.updateRefs(builder.build());
        }
    }
}
