package com.atlassian.bitbucket.mesh.git;

import com.atlassian.bitbucket.mesh.AbstractStatusException;
import com.atlassian.bitbucket.mesh.coordination.LogicalClock;
import com.atlassian.bitbucket.mesh.coordination.WriteTokenRegistry;
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.network.MeshManager;
import com.atlassian.bitbucket.mesh.repository.*;
import com.atlassian.bitbucket.mesh.rpc.v1.*;
import com.atlassian.bitbucket.mesh.transaction.RpcTransactionExecutor;
import com.atlassian.bitbucket.mesh.transaction.TransactionManager;
import com.atlassian.bitbucket.mesh.util.ConcurrencyUtils;
import com.google.common.util.concurrent.MoreExecutors;
import io.grpc.stub.StreamObserver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import static com.atlassian.bitbucket.mesh.MeshConstants.EXECUTOR_TRANSACTIONS;

public class GrpcTransactionService extends TransactionServiceGrpc.TransactionServiceImplBase {

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

    private final GitProcessBuilderFactory builderFactory;
    private final EventPublisher eventPublisher;
    private final ExecutionManager executionManager;
    private final ExecutorService executorService;
    private final LogicalClock logicalClock;
    private final MeshManager meshManager;
    private final ReplicaStateRegistry replicaStateRegistry;
    private final RepositoryAttributeManager repositoryAttributeManager;
    private final RepositoryHashService repositoryHashService;
    private final RepositoryManager repositoryManager;
    private final RpcTransactionExecutor transactionExecutor;
    private final TransactionManager transactionManager;
    private final WriteTokenRegistry writeTokenRegistry;

    public GrpcTransactionService(GitProcessBuilderFactory builderFactory, EventPublisher eventPublisher,
                                  ExecutionManager executionManager, LogicalClock logicalClock, MeshManager meshManager,
                                  ReplicaStateRegistry replicaStateRegistry,
                                  RepositoryAttributeManager repositoryAttributeManager,
                                  RepositoryHashService repositoryHashService, RepositoryManager repositoryManager,
                                  RpcTransactionExecutor transactionExecutor, TransactionManager transactionManager,
                                  WriteTokenRegistry writeTokenRegistry) {
        this.builderFactory = builderFactory;
        this.eventPublisher = eventPublisher;
        this.executionManager = executionManager;
        this.logicalClock = logicalClock;
        this.meshManager = meshManager;
        this.replicaStateRegistry = replicaStateRegistry;
        this.repositoryAttributeManager = repositoryAttributeManager;
        this.repositoryHashService = repositoryHashService;
        this.repositoryManager = repositoryManager;
        this.transactionExecutor = transactionExecutor;
        this.transactionManager = transactionManager;
        this.writeTokenRegistry = writeTokenRegistry;

        executorService = Executors.newCachedThreadPool(
                ConcurrencyUtils.createDaemonThreadFactory(EXECUTOR_TRANSACTIONS));
    }

    @Override
    public void applyTransactions(RpcApplyTransactionsRequest request,
                                  StreamObserver<RpcApplyTransactionsResponse> responseObserver) {
        int processed = 0;
        for (RpcTransaction transaction : request.getTransactionsList()) {
            try {
                transactionManager.executeWithoutResult(tx -> transactionExecutor.execute(transaction));
                ++processed;
            } catch (RepositoryAlreadyExistsException e) {
                //When applying catch-up transactions, it's not an error if the repository already exists
                log.info("[{}] Repository already exists. Skipping replay of repository creation transaction", e.getId());
            } catch (RuntimeException e) {
                log.error("Failed to apply transaction {}: {}", processed, transaction, e);

                responseObserver.onNext(buildApplyResponse(processed, e));
                responseObserver.onCompleted();
                return;
            }
        }
        responseObserver.onNext(buildApplyResponse(processed, null));
        responseObserver.onCompleted();
    }

    public void shutdown() {
        if (MoreExecutors.shutdownAndAwaitTermination(executorService, 10L, TimeUnit.SECONDS)) {
            log.debug("The transaction executor has shutdown");
        } else {
            log.warn("Failed to shutdown the transaction executor; there may be active transactions");
        }
    }

    @Nonnull
    @Override
    public StreamObserver<RpcTransactLeaderFragment> transact(
            StreamObserver<RpcTransactFollowerFragment> responseObserver) {
        return new RemoteTransactionLeader(responseObserver, builderFactory, eventPublisher, executionManager,
                executorService, logicalClock, meshManager, replicaStateRegistry, repositoryAttributeManager,
                repositoryHashService, repositoryManager, transactionExecutor, transactionManager, writeTokenRegistry);
    }

    @Override
    public StreamObserver<RpcVerifyConsistencyLeaderFragment> verifyConsistency(
            StreamObserver<RpcVerifyConsistencyFollowerFragment> responseObserver) {
        return new VerifyConsistencyFollower(responseObserver, eventPublisher, executionManager, logicalClock,
                meshManager, repositoryAttributeManager, repositoryHashService, repositoryManager, writeTokenRegistry);
    }

    private static RpcApplyTransactionsResponse buildApplyResponse(int successCount, RuntimeException e) {
        RpcApplyTransactionsResponse.Builder builder = RpcApplyTransactionsResponse.newBuilder()
                .setSuccessCount(successCount);
        if (e instanceof AbstractStatusException) {
            e = ((AbstractStatusException) e).toStatusException();
        }
        if (e != null) {
            RpcGitError gitError = RpcGitErrorUtils.fromThrowable(e);
            if (gitError == null) {
                RpcError error = RpcErrorUtils.fromThrowable(e);
                if (error == null) {
                    builder.setError(RpcError.newBuilder()
                            .setTransactionApplyError(RpcTransactionApplyError.newBuilder()
                                    .setMessage(e.getMessage())));
                } else {
                    builder.setError(error);
                }
            } else {
                builder.setGitError(gitError);
            }
        }
        return builder.build();
    }
}
