package com.atlassian.bitbucket.mesh.client;

import com.atlassian.bitbucket.mesh.config.ApplicationConfig;
import com.atlassian.bitbucket.mesh.coordination.LogicalClock;
import com.atlassian.bitbucket.mesh.event.EventPublisher;
import com.atlassian.bitbucket.mesh.git.GitPackFile;
import com.atlassian.bitbucket.mesh.git.builder.GitProcessBuilderFactory;
import com.atlassian.bitbucket.mesh.git.gc.GarbageCollectionManager;
import com.atlassian.bitbucket.mesh.git.porcelain.GitPorcelain;
import com.atlassian.bitbucket.mesh.git.transcode.TranscodeService;
import com.atlassian.bitbucket.mesh.network.ChannelRegistry;
import com.atlassian.bitbucket.mesh.network.ConnectivitySummary;
import com.atlassian.bitbucket.mesh.network.MeshNode;
import com.atlassian.bitbucket.mesh.repair.InteractiveRepairHelper;
import com.atlassian.bitbucket.mesh.repair.RepairSource;
import com.atlassian.bitbucket.mesh.repair.RepairSummary;
import com.atlassian.bitbucket.mesh.repair.RepairTarget;
import com.atlassian.bitbucket.mesh.repository.ReplicaStateRegistry;
import com.atlassian.bitbucket.mesh.repository.Repository;
import com.atlassian.bitbucket.mesh.repository.RepositoryHashService;
import com.atlassian.bitbucket.mesh.repository.RepositoryManager;
import com.atlassian.bitbucket.mesh.rpc.v1.*;
import com.atlassian.bitbucket.mesh.rpc.v1.ManagementServiceGrpc.ManagementServiceStub;
import com.atlassian.bitbucket.mesh.rpc.v1.PlumbingServiceGrpc.PlumbingServiceStub;
import com.atlassian.bitbucket.mesh.rpc.v1.RepositoryServiceGrpc.RepositoryServiceStub;
import com.atlassian.bitbucket.mesh.rpc.v1.TransactionServiceGrpc.TransactionServiceStub;
import com.atlassian.bitbucket.mesh.transaction.ReplicaConsistencyVerifier;
import com.atlassian.bitbucket.mesh.transaction.ReplicaTransaction;
import com.google.protobuf.ByteString;

import javax.annotation.Nonnull;
import java.time.Duration;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

import static com.google.common.base.Preconditions.checkArgument;
import static java.util.Objects.requireNonNull;

public class GrpcMeshClient implements MeshClient {

    private final GitProcessBuilderFactory builderFactory;
    private final ChannelRegistry channelRegistry;
    private final ApplicationConfig config;
    private final EventPublisher eventPublisher;
    private final ExecutorService executorService;
    private final GarbageCollectionManager garbageCollectionManager;
    private final LogicalClock logicalClock;
    private final GitPorcelain porcelain;
    private final InteractiveRepairHelper repairHelper;
    private final ReplicaStateRegistry replicaStateRegistry;
    private final RepositoryHashService repositoryHashService;
    private final RepositoryManager repositoryManager;
    private final TranscodeService transcodeService;

    public GrpcMeshClient(GitProcessBuilderFactory builderFactory, ChannelRegistry channelRegistry,
                          ApplicationConfig config, EventPublisher eventPublisher, ExecutorService executorService,
                          GarbageCollectionManager garbageCollectionManager, LogicalClock logicalClock,
                          GitPorcelain porcelain, InteractiveRepairHelper repairHelper,
                          ReplicaStateRegistry replicaStateRegistry, RepositoryHashService repositoryHashService,
                          RepositoryManager repositoryManager, TranscodeService transcodeService) {
        this.builderFactory = builderFactory;
        this.channelRegistry = channelRegistry;
        this.config = config;
        this.eventPublisher = eventPublisher;
        this.executorService = executorService;
        this.garbageCollectionManager = garbageCollectionManager;
        this.logicalClock = logicalClock;
        this.porcelain = porcelain;
        this.repairHelper = repairHelper;
        this.replicaStateRegistry = replicaStateRegistry;
        this.repositoryHashService = repositoryHashService;
        this.repositoryManager = repositoryManager;
        this.transcodeService = transcodeService;
    }

    @Nonnull
    @Override
    public GitPackFile fetchObjects(@Nonnull MeshNode node, @Nonnull Repository repository,
                                    @Nonnull FetchObjectsParameters parameters) {
        requireNonNull(node, "node");
        requireNonNull(repository, "repository");
        requireNonNull(parameters, "parameters");

        PackObjectsResponseObserver responseObserver = new PackObjectsResponseObserver(
                builderFactory, node, repository, parameters);

        PlumbingServiceStub stub = channelRegistry.getChannel(node).newStub(PlumbingServiceGrpc::newStub);
        stub.packObjects(buildPackObjectsRequest(repository, parameters), responseObserver);

        return new GitPackFile(responseObserver.asResult(), parameters.isKeep());
    }

    @Nonnull
    @Override
    public ConnectivitySummary ping(@Nonnull MeshNode node) {
        requireNonNull(node, "node");

        ConnectivitySummaryResponseObserver responseObserver = new ConnectivitySummaryResponseObserver();
        ManagementServiceStub stub = channelRegistry.getChannel(node).newStub(ManagementServiceGrpc::newStub);
        stub.withDeadlineAfter(1L, TimeUnit.MINUTES)
                .heartbeat(RpcHeartbeat.getDefaultInstance(), responseObserver);

        return responseObserver.asResult();
    }

    @Nonnull
    @Override
    public RepairSummary repair(@Nonnull MeshNode node, @Nonnull RepairRepositoryParameters parameters) {
        requireNonNull(node, "node");
        requireNonNull(parameters, "parameters");

        Duration timeout = Duration.ofSeconds(parameters.getExecutionTimeout());
        RepairSource observer = new RepairSource(builderFactory, executorService, garbageCollectionManager,
                config.getGrpc(), logicalClock, porcelain, replicaStateRegistry, repositoryHashService,
                repositoryManager, transcodeService, timeout);
        RepositoryServiceStub stub = channelRegistry.getChannel(node).newStub(RepositoryServiceGrpc::newStub);
        stub.withDeadlineAfter(timeout.toMillis(), TimeUnit.MILLISECONDS).repair(observer);

        observer.startRepair(parameters.getSourceRepositoryId(), parameters.getRepositoryId());
        return observer.asResult();
    }

    @Nonnull
    @Override
    public RepairSummary repairMe(@Nonnull MeshNode node, @Nonnull RepairRepositoryParameters parameters) {
        requireNonNull(node, "node");
        requireNonNull(parameters, "parameters");

        Duration timeout = Duration.ofSeconds(parameters.getExecutionTimeout());
        RepairTarget observer = new RepairTarget(eventPublisher, config.getGrpc(), logicalClock, repairHelper, timeout);
        RepositoryServiceStub stub = channelRegistry.getChannel(node).newStub(RepositoryServiceGrpc::newStub);
        stub.withDeadlineAfter(timeout.toMillis(), TimeUnit.MILLISECONDS).repairMe(observer);

        observer.startRepair(parameters.getSourceRepositoryId(), parameters.getRepositoryId());
        return observer.asResult();
    }

    @Nonnull
    @Override
    public ReplicaTransaction replicate(@Nonnull MeshNode node, @Nonnull ReplicateParameters parameters) {
        checkArgument(requireNonNull(node, "node").isRemote(), "%s is not a remote node", node);
        requireNonNull(parameters, "parameters");

        ReplicaTransactionResponseObserver observer =
                new ReplicaTransactionResponseObserver(logicalClock, node, parameters);

        TransactionServiceStub stub = channelRegistry.getChannel(node).newStub(TransactionServiceGrpc::newStub);
        stub.transact(observer);

        return observer;
    }

    @Nonnull
    @Override
    public ReplicaConsistencyVerifier verifyConsistency(@Nonnull MeshNode node, @Nonnull ReplicateParameters parameters) {
        checkArgument(requireNonNull(node, "node").isRemote(), "%s is not a remote node", node);
        requireNonNull(parameters, "parameters");

        ReplicaConsistencyResponseObserver observer =
                new ReplicaConsistencyResponseObserver(logicalClock, node, parameters);

        TransactionServiceStub stub = channelRegistry.getChannel(node).newStub(TransactionServiceGrpc::newStub);
        stub.verifyConsistency(observer);

        return observer;
    }

    private static RpcPackObjectsRequest buildPackObjectsRequest(Repository repository,
                                                                 FetchObjectsParameters parameters) {
        RpcPackObjectsRequest.Builder builder = RpcPackObjectsRequest.newBuilder()
                .setDeltaBaseOffset(true)
                .setRepository(repository.getId())
                .setThin(true)
                .setTimeouts(RpcGitTimeouts.newBuilder()
                        .setExecution(parameters.getExecutionTimeout())
                        .setIdle(parameters.getIdleTimeout()));
        parameters.getIncludes().forEach(include -> builder.addRevs(ByteString.copyFromUtf8(include)));
        parameters.getExcludes().forEach(exclude -> builder.addRevs(ByteString.copyFromUtf8("^" + exclude)));
        return builder.build();
    }
}
