package com.atlassian.bitbucket.mesh;

import com.atlassian.bitbucket.mesh.execution.ExecutionManager;
import com.atlassian.bitbucket.mesh.git.GitAgent;
import com.atlassian.bitbucket.mesh.git.GitObjectType;
import com.atlassian.bitbucket.mesh.git.builder.GitProcessBuilderFactory;
import com.atlassian.bitbucket.mesh.git.builder.checkout.GitCheckoutCreateBranchMode;
import com.atlassian.bitbucket.mesh.git.builder.fetch.GitFetchBuilder;
import com.atlassian.bitbucket.mesh.git.builder.updateref.GitUpdateRefBuilderSupport;
import com.atlassian.bitbucket.mesh.git.exception.CommandFailedException;
import com.atlassian.bitbucket.mesh.git.ref.*;
import com.atlassian.bitbucket.mesh.grpc.AbstractFutureResponseObserver;
import com.atlassian.bitbucket.mesh.io.MoreFiles;
import com.atlassian.bitbucket.mesh.junit.TemporaryDirectory;
import com.atlassian.bitbucket.mesh.network.ChannelRegistry;
import com.atlassian.bitbucket.mesh.network.MeshManager;
import com.atlassian.bitbucket.mesh.network.MeshNode;
import com.atlassian.bitbucket.mesh.process.handler.LoggingStdioHandler;
import com.atlassian.bitbucket.mesh.process.handler.StringStdioHandler;
import com.atlassian.bitbucket.mesh.repair.RepairCapability;
import com.atlassian.bitbucket.mesh.repository.*;
import com.atlassian.bitbucket.mesh.rpc.v1.*;
import com.atlassian.bitbucket.mesh.rpc.v1.ManagementServiceGrpc.ManagementServiceBlockingStub;
import com.atlassian.bitbucket.mesh.rpc.v1.PlumbingServiceGrpc.PlumbingServiceStub;
import com.atlassian.bitbucket.mesh.rpc.v1.RefServiceGrpc.RefServiceBlockingStub;
import com.atlassian.bitbucket.mesh.rpc.v1.RepositoryServiceGrpc.RepositoryServiceBlockingStub;
import com.atlassian.bitbucket.mesh.rpc.v1.RepositoryServiceGrpc.RepositoryServiceStub;
import com.atlassian.bitbucket.mesh.test.cluster.MeshClusterManager;
import com.atlassian.bitbucket.mesh.util.ShaUtils;
import com.google.protobuf.ByteString;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.test.context.TestContext;
import org.springframework.test.context.support.AbstractTestExecutionListener;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.io.ByteArrayOutputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.nio.file.Files;
import java.nio.file.NoSuchFileException;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.attribute.FileTime;
import java.time.Duration;
import java.time.Instant;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Supplier;
import java.util.stream.Stream;

import static com.atlassian.bitbucket.mesh.git.GitConstants.PATH_APPLICATION_INFO;
import static com.atlassian.bitbucket.mesh.io.MoreFiles.getLastModified;
import static io.atlassian.fugue.Iterables.filter;
import static io.atlassian.fugue.Iterables.map;
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Objects.requireNonNull;
import static java.util.stream.Collectors.toList;

public class TestHelper extends AbstractTestExecutionListener {

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

    private final GitAgent agent;
    private final GitProcessBuilderFactory builderFactory;
    private final ChannelRegistry channelRegistry;
    private final MeshClusterManager clusterManager;
    private final Map<MeshNode, Map<String, String>> configToRestore;
    private final ExecutionManager executionManager;
    private final MeshManager meshManager;
    private final RepositoryManager repositoryManager;

    public TestHelper(GitAgent agent, GitProcessBuilderFactory builderFactory,
                      ChannelRegistry channelRegistry, MeshClusterManager clusterManager,
                      ExecutionManager executionManager, MeshManager meshManager,
                      RepositoryManager repositoryManager) {
        this.agent = agent;
        this.builderFactory = builderFactory;
        this.channelRegistry = channelRegistry;
        this.clusterManager = clusterManager;
        this.executionManager = executionManager;
        this.meshManager = meshManager;
        this.repositoryManager = repositoryManager;
        configToRestore = new ConcurrentHashMap<>();
    }

    @Override
    public void afterTestMethod(@Nonnull TestContext testContext) {
        configToRestore.forEach((node, configMap) -> {
            blockingManagementStub(node).setConfiguration(RpcSetConfigurationRequest.newBuilder()
                    .setNodeConfiguration(RpcNodeConfiguration.newBuilder().putAllProperties(configMap)).build());
        });
        configToRestore.clear();
    }

    public void assertRefsInSync(@Nonnull Collection<MeshNode> nodes, @Nonnull String repositoryId) {
        Map<String, List<MeshNode>> nodesByForEachRefOutput = new HashMap<>();
        RpcForEachRefRequest request = RpcForEachRefRequest.newBuilder()
                .setRepository(repositoryId)
                .setFormat("%(objectname) %(refname)")
                .build();
        for (MeshNode node : nodes) {
            StringStreamObserver<RpcForEachRefRequest> observer = new StringStreamObserver<>();
            plumbingStub(node).forEachRef(request, observer);
            nodesByForEachRefOutput.computeIfAbsent(observer.asResult(), ignored -> new ArrayList<>()).add(node);
        }

        if (nodesByForEachRefOutput.size() != 1) {
            throw new IllegalStateException("Not all nodes have the same refs: " + nodesByForEachRefOutput);
        }
    }

    public Branch createBranchWithNewCommit(@Nonnull MeshNode node, @Nonnull String repositoryId,
                                            @Nonnull String refId) {
        //Create the new object in the repository replica
        String commitId = createCommit(node, repositoryId);
        //Create the branch through the RefService/UpdateRef RPC to trigger replication
        blockingRefStub(node).updateRef(RpcUpdateRefRequest.newBuilder()
                .setRepository(repositoryId)
                .setRefUpdate(RpcProposedRefUpdate.newBuilder()
                        .setRefId(ByteString.copyFromUtf8(refId))
                        .setNewHash(commitId)
                        .setOldHash(ShaUtils.NULL_SHA1))
                .build());

        return new Branch.Builder()
                .id(refId)
                .latestCommit(commitId)
                .build();
    }

    /**
     * Creates a commit that introduces a new file. Then adds the commit to the repository replica's objects directory
     * on the provided node.
     *
     * @param node         optional node; {@code null} for the local node
     * @param repositoryId the repository ID
     * @return the commit ID
     * @throws UncheckedIOException if an I/O error occurred while preparing the commit
     */
    @Nonnull
    public String createCommit(@Nullable MeshNode node,
                               @Nonnull String repositoryId) {
        requireNonNull(node, "node");
        requireNonNull(repositoryId, "repositoryId");

        TemporaryDirectory temporaryDirectory = new TemporaryDirectory();
        try {
            temporaryDirectory.create();

            Path repoDir = getRepoDir(node, repositoryId);
            Path cloneDir = temporaryDirectory.getRoot().resolve("clone");

            //Create a clone
            builderFactory.builder().clone()
                    .normal()
                    .origin(repoDir)
                    .directory(cloneDir)
                    .build()
                    .run();

            //Create a commit on another branch
            String random = UUID.randomUUID().toString();
            String refId = "refs/heads/" + random;
            builderFactory.builder().checkout()
                    .branch(random, null, GitCheckoutCreateBranchMode.CREATE)
                    .workDir(cloneDir)
                    .build(new LoggingStdioHandler())
                    .run();
            Path newFile = cloneDir.resolve(random);
            MoreFiles.write(newFile, "New content");
            builderFactory.builder().add()
                    .path(newFile.toString())
                    .workDir(cloneDir)
                    .build()
                    .run();
            builderFactory.builder().commit()
                    .author("Michael Heemskerk", "mheemskerk@atlassian.com")
                    .message("Time for a change")
                    .workDir(cloneDir)
                    .build()
                    .run();
            Branch branch = agent.resolveHead(new DefaultPathRepository(cloneDir.resolve(".git")));

            fetch(cloneDir, repoDir, refId);
            return branch.getLatestCommit();
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        } finally {
            temporaryDirectory.delete();
        }
    }

    public void createLooseRef(MeshNode node, Repository repository, String refId, String hash) {
        Path refPath = getRepoDir(node, repository.getId()).resolve(refId);
        MoreFiles.mkdir(refPath.getParent());
        try {
            MoreFiles.write(refPath, hash);
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
    }

    /**
     * Creates a new repository on the provided node.
     *
     * @param node       the node to send the create request to
     * @param parameters parameters describing the repository to create
     */
    public void createRepository(@Nonnull MeshNode node, @Nonnull CreateRepositoryParameters parameters) {
        RpcCreateRepositoryRequest.Builder builder = RpcCreateRepositoryRequest.newBuilder()
                .putAllMetadata(parameters.getMetadata())
                .setRepository(parameters.getId());

        String defaultBranch = parameters.getDefaultBranch();
        if (defaultBranch != null) {
            builder.setDefaultBranch(ByteString.copyFromUtf8(defaultBranch));
        }

        blockingRepositoryStub(node).create(builder.build());
    }

    /**
     * Deletes the repository identified by {@code repositoryId} on the provided node.
     *
     * @param node         the node to send the delete request to
     * @param repositoryId the repository to delete
     */
    public void deleteRepository(@Nonnull MeshNode node, @Nonnull String repositoryId) {
        blockingRepositoryStub(node).delete(RpcDeleteRepositoryRequest.newBuilder()
                .setRepository(repositoryId)
                .build());
    }

    /**
     * Deletes the repository identified by {@code repositoryId} on the provided node.
     *
     * @param node         the node to send the delete request to
     * @param repositoryId the repository to delete
     */
    public void deleteRepositoryWithoutCoordination(@Nonnull MeshNode node, @Nonnull String repositoryId) {
        if (ManagedRepository.Id.matches(repositoryId)) {
            Path repoDir = getRepoDir(node, repositoryId);
            MoreFiles.deleteQuietly(repoDir);
            try {
                if (Files.list(repoDir.getParent()).count() <= 1) {
                    //Only repo in the hierarchy; delete the whole hierarchy dir
                    MoreFiles.deleteQuietly(repoDir.getParent().getParent());
                }
            } catch (IOException e) {
                //Ignore
            }
        }
    }

    /**
     * Fetches objects identified by the provided refspec into to repository replica on the provided node.
     *
     * @param node         the node holding the repository replica to update
     * @param repositoryId the repository ID
     * @param from         the repository directory to fetch from
     * @param refspecs     describes the branches/objects to fetch
     */
    public void fetch(@Nonnull MeshNode node, @Nonnull String repositoryId, @Nonnull Path from,
                      @Nonnull String... refspecs) {
        Path repoDir = getRepoDir(node, repositoryId);
        fetch(from, repoDir, refspecs);
    }

    /**
     * Retrieves the default branch for the provided repository from the provided node.
     *
     * @param node         the node to query
     * @param repositoryId the repository ID
     * @return the default branch
     */
    @Nonnull
    public String getDefaultBranch(@Nonnull MeshNode node, @Nonnull String repositoryId) {
        return blockingRepositoryStub(node).getDefaultBranch(RpcGetDefaultBranchRequest.newBuilder()
                .setRepository(repositoryId)
                .build())
                .getBranchId()
                .toStringUtf8();
    }

    /**
     * Retrieves details for the provided repository (including default branch, metadata, etc.) from the provided node.
     *
     * @param node         the node to query
     * @param repositoryId the repository ID
     * @return the repository detail
     */
    @Nonnull
    public RpcGetRepositoryDetailsResponse getDetails(@Nonnull MeshNode node, @Nonnull String repositoryId) {
        return blockingRepositoryStub(node).getDetails(RpcGetRepositoryDetailsRequest.newBuilder()
                .setRepository(repositoryId)
                .build());
    }

    @Nullable
    public String getLooseRef(MeshNode node, Repository repository, String refId) {
        Path refPath = getRepoDir(node, repository.getId()).resolve(refId);
        try {
            return MoreFiles.toString(refPath).trim();
        } catch (FileNotFoundException | NoSuchFileException e) {
            return null;
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
    }

    /**
     * @param node       optional node to query, or {@code null} for the local node
     * @param repository the repository to query
     * @param objectId   the ID of the object to check
     * @return the object type, or {@code null} if the object was not found
     */
    @Nullable
    public GitObjectType getObjectType(@Nullable MeshNode node, @Nonnull ManagedRepository repository,
                                       @Nonnull String objectId) {
        requireNonNull(repository, "repository");
        requireNonNull(objectId, "objectId");

        try {
            return GitObjectType.fromObjectType(builderFactory.builder().catFile()
                    .type()
                    .object(objectId)
                    .workDir(getRepoDir(node, repository))
                    .build(new StringStdioHandler())
                    .run());
        } catch (CommandFailedException e) {
            return null;
        }
    }

    /**
     * @param node       optional node to query, or {@code null} for the local node
     * @param repository the repository to query
     * @param refId      the ref for which the reflog should be retrieved
     * @return           the reflog contents, or {@code null} if the reflog does not exist
     */
    @Nullable
    public String getReflog(@Nullable MeshNode node, @Nonnull ManagedRepository repository, @Nonnull String refId) {
        Path reflogPath = getRepoDir(node, repository).resolve("logs").resolve(refId);
        try {
            return MoreFiles.toString(reflogPath, UTF_8);
        } catch (FileNotFoundException | NoSuchFileException e) {
            return null;
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
    }

    /**
     * Returns an iterable of reflog paths, including directories, relative to the reflog directory and sorted by name.
     *
     * @param node       the node to get the reflogs from
     * @param repository the repository to get the reflogs for
     * @return the iterable of reflog paths
     */
    public Iterable<Path> getReflogs(@Nullable MeshNode node, @Nonnull Repository repository) {
        Path repoDir = getRepoDir(node, repository);
        Path reflogPath = repoDir.resolve("logs");
        return filter(
                map(() -> MoreFiles.iterateFileTreeSorted(reflogPath),
                        path -> reflogPath.relativize(path.getPath())),
                path -> !path.toString().isEmpty());
    }

    @Nonnull
    public RpcGetRepositoryStateResponse getReplicaState(@Nonnull MeshNode node, @Nonnull String repositoryId) {
        AbstractFutureResponseObserver<RpcRepairTargetFragment, RpcRepairSourceFragment, RpcGetRepositoryStateResponse> observer =
                new AbstractFutureResponseObserver<RpcRepairTargetFragment, RpcRepairSourceFragment, RpcGetRepositoryStateResponse>() {

                    @Override
                    public void onNext(RpcRepairSourceFragment value) {
                        if (value.hasRepositoryState()) {
                            setResult(value.getRepositoryState());
                        }
                    }

                    @Override
                    public RpcGetRepositoryStateResponse asResult() {
                        requestStream.onNext(RpcRepairTargetFragment.newBuilder()
                                .setGetRepositoryState(RpcGetRepositoryStateRequest.newBuilder()
                                        .setCapabilities(RepairCapability.ADVERTISEMENT)
                                        .setRepository(repositoryId))
                                .build());
                        requestStream.onCompleted();
                        return super.asResult();
                    }
                };
        repositoryStub(node).repairMe(observer);
        return observer.asResult();
    }

    public boolean isMarkedDirty(@Nonnull MeshNode node, @Nonnull String repositoryId) {
        Path repoDir = getRepoDir(node, repositoryId);
        Path dirtyMarker = repoDir.getParent().resolve("app-info/dirty").resolve(repoDir.getFileName());

        return Files.isRegularFile(dirtyMarker);
    }

    /**
     * @param node         the node to query
     * @param repositoryId the repository to query
     * @return {@code true} if transcoding is enabled for the repository on the provided node
     */
    public boolean isTranscodeEnabled(@Nonnull MeshNode node, @Nonnull String repositoryId) {
        return getDetails(node, repositoryId).getTranscodeEnabled();
    }

    @Nonnull
    public ReplicaState repair(@Nonnull MeshNode node, @Nonnull MeshNode repairSource,
                               @Nonnull MeshNode repairTarget, @Nonnull String repositoryId) {
        requireNonNull(node, "node");
        requireNonNull(repairSource, "repairSource");
        requireNonNull(repairTarget, "repairTarget");
        requireNonNull(repositoryId, "repositoryId");

        RpcRepairRepositoryRequest.Builder builder = RpcRepairRepositoryRequest.newBuilder()
                .setRepository(repositoryId);
        if (node.equals(repairSource)) {
            builder.setTarget(repairTarget.getId());
        } else {
            builder.setSources(RpcNodeIdList.newBuilder().addNodes(repairSource.getId()));
        }

        return ReplicaState.fromRpc(blockingManagementStub(node).repairRepository(builder.build()).getState());
    }

    /**
     * Resolves the provided {@code refId} in the repository replica located on the provided node.
     *
     * @param node         the node to query
     * @param repositoryId the repository to query
     * @param refId        the ref ID to resolve
     * @return the resolved ref, or {@code null} if the ref does not exist
     */
    public Ref resolveRef(@Nonnull MeshNode node, @Nonnull String repositoryId, @Nonnull String refId) {
        RpcResolveRefsResponse response = blockingRefStub(node).resolveRefs(RpcResolveRefsRequest.newBuilder()
                .addRefIds(ByteString.copyFromUtf8(refId))
                .setRepository(repositoryId)
                .build());

        if (response.getResolvedRefsCount() == 0) {
            return null;
        }
        RpcResolvedRef ref = response.getResolvedRefs(0);
        if (ref.hasBranch()) {
            return Branch.fromBranch(ref.getBranch());
        }
        if (ref.hasTag()) {
            return Tag.fromTag(ref.getTag());
        }
        return null;
    }

    /**
     * Schedule garbage collection and vacuum on the specified node. This returns a {@link Supplier} that returns
     * {@code true} when garbage collection has finished running. Use code like this to wait for it:
     *
     * <pre>{@code
     * awaitUntil(runGarbageCollectionNow(node, repository), is(true), Duration.ofSeconds(120L), Duration.ofSeconds(1L));
     * }</pre>
     *
     * @param node       the node to run garbage collection on
     * @param repository the repository to run garbage collection for
     * @return a supplier that returns {@code true} when garbage collection has finished running
     */
    public Supplier<Boolean> scheduleGcAndVacuum(MeshNode node, ManagedRepository repository) {
        return scheduleGcAndVacuum(node, repository, false);
    }

    /**
     * Schedule garbage collection and vacuum on the specified node. This returns a {@link Supplier} that returns
     * {@code true} when garbage collection has finished running. Use code like this to wait for it:
     *
     * <pre>{@code
     * awaitUntil(runGarbageCollectionNow(node, repository), is(true), Duration.ofSeconds(120L), Duration.ofSeconds(1L));
     * }</pre>
     *
     * @param node       the node to run garbage collection on
     * @param repository the repository to run garbage collection for
     * @param fullVacuum
     * @return a supplier that returns {@code true} when garbage collection has finished running
     */
    public Supplier<Boolean> scheduleGcAndVacuum(MeshNode node, ManagedRepository repository, boolean fullVacuum) {
        Instant ts = Instant.now().minus(Duration.ofDays(365));
        Instant now = Instant.now().minus(Duration.ofSeconds(5L));
        Path appInfo = Paths.get(PATH_APPLICATION_INFO);
        Stream<String> filesToCheck = Stream.of("packrefs.ts", "vacuum.ts");
        if (fullVacuum) {
            filesToCheck = Stream.concat(filesToCheck, Stream.of("vacuum-full.ts"));
        }
        List<Path> paths = filesToCheck.map(appInfo::resolve).collect(toList());
        paths.forEach(path -> touchFile(node, repository, path, ts));
        blockingRepositoryStub(node).scheduleGarbageCollection(RpcScheduleGarbageCollectionRequest.newBuilder()
                .setRepository(repository.getId())
                .setStaleOnly(false)
                .build());
        return () -> !paths.stream().map(getRepoDir(node, repository)::resolve)
                .map(path -> Pair.of(path, Instant.ofEpochMilli(getLastModified(path))))
                .filter(pair -> now.isAfter(pair.getRight()))
                .peek(pair -> log.debug("Waiting for {} to update", pair.getLeft()))
                .findAny().isPresent();
    }

    /**
     * Set the specified configuration on the node.
     * <p>
     * An attempt is made to restore the default value after the test execution has finished.
     *
     * @param config the config property name
     * @param value  the new value
     */
    public void setConfig(String config, String value) {
        meshManager.getNodes().forEach(node -> setConfig(node, config, value));
    }

    /**
     * Set the specified configuration on the node.
     * <p>
     * An attempt is made to restore the default value after the test execution has finished.
     *
     * @param node   the node to update the configuration on
     * @param config the config property name
     * @param value  the new value
     */
    public void setConfig(MeshNode node, String config, String value) {
        configToRestore.computeIfAbsent(node, (ignored) -> new ConcurrentHashMap<>())
                .computeIfAbsent(config, (ignored) ->
                        blockingManagementStub(node).getConfiguration(RpcGetConfigurationRequest.newBuilder().build())
                                .getNodeConfiguration()
                                .getPropertiesMap()
                                .get(config));
        blockingManagementStub(node).setConfiguration(RpcSetConfigurationRequest.newBuilder()
                .setNodeConfiguration(RpcNodeConfiguration.newBuilder()
                        .putProperties(config, value))
                .build());
    }

    /**
     * Updates the default branch for the provided repository replica
     *
     * @param node         the node that hosts the repository replica
     * @param repositoryId the repository
     * @param value        the new default branch value
     */
    public void setDefaultBranch(@Nonnull MeshNode node, @Nonnull String repositoryId, @Nonnull String value) {
        blockingRepositoryStub(node).setDefaultBranch(RpcSetDefaultBranchRequest.newBuilder()
                .setBranchId(ByteString.copyFromUtf8(value))
                .setRepository(repositoryId)
                .build());
    }

    public void setReadOnly(@Nonnull MeshNode node, @Nonnull String repositoryId, boolean value) {
        blockingManagementStub(node).setRepositoryReadOnly(RpcSetRepositoryReadOnlyRequest.newBuilder()
                .setRepository(repositoryId)
                .setReadOnly(value)
                .build());
    }

    /**
     * Updates the reflog for the provided ref in the provided repository replica.
     *
     * @param node       the node hosting the repository replica to update
     * @param repository the repository to update
     * @param refId      the ref for which the reflog should be updated
     * @param contents   the new reflog contents; {@code null} to delete the reflog
     */
    public void setReflog(@Nullable MeshNode node, @Nonnull Repository repository, @Nonnull String refId,
                          @Nullable String contents, @Nullable Instant modificationTimestamp) {
        Path repoDir = getRepoDir(node, repository);
        Path reflogPath = repoDir.resolve("logs").resolve(refId);
        try {
            if (contents == null) {
                Files.deleteIfExists(reflogPath);
            } else {
                MoreFiles.mkdir(reflogPath.getParent());
                MoreFiles.write(reflogPath, contents, UTF_8);
                if (modificationTimestamp != null) {
                    Files.setLastModifiedTime(reflogPath, FileTime.from(modificationTimestamp));
                }
            }
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
    }

    /**
     * Updates the reflog for the provided ref in the provided repository replica.
     *
     * @param node       the node hosting the repository replica to update
     * @param repository the repository to update
     * @param refId      the ref for which the reflog should be updated
     * @param contents   the new reflog contents; {@code null} to delete the reflog
     */
    public void setReflog(@Nullable MeshNode node, @Nonnull Repository repository, @Nonnull String refId,
                          @Nullable String contents) {
        setReflog(node, repository, refId, contents, null);
    }

    @Nonnull
    public void setReplicaState(@Nonnull MeshNode node, @Nonnull String repositoryId,
                                @Nonnull ReplicaState replicaState) {
        RpcGetRepositoryStateResponse state = getReplicaState(node, repositoryId);
        blockingManagementStub(node).setReplicaState(RpcSetReplicaStateRequest.newBuilder()
                .setReplicaState(replicaState.toRpc())
                .setRepository(repositoryId)
                .setVersion(state.getVersion())
                .setForce(true)
                .build());
    }

    /**
     * Updates the diff transcoding setting for the provided repository replica.
     *
     * @param node         the node hosting the repository replica to update
     * @param repositoryId the repository
     * @param value        the new transcoding setting
     */
    public void setTranscodeEnabled(@Nonnull MeshNode node, @Nonnull String repositoryId, boolean value) {
        blockingRepositoryStub(node).setTranscodeEnabled(RpcSetTranscodeEnabledRequest.newBuilder()
                .setEnabled(value)
                .setRepository(repositoryId)
                .build());
    }

    /**
     * Updates the ref for the repository replica as specified by {@code parameters}.
     *
     * @param node       the node hosting the repository replica to update
     * @param parameters describes the ref to update
     */
    public void updateRef(@Nonnull MeshNode node, @Nonnull UpdateRefParameters parameters) {
        blockingRefStub(node).updateRef(RpcUpdateRefRequest.newBuilder()
                .setRepository(parameters.getRepository().getId())
                .setRefUpdate(parameters.getUpdate().toRefUpdate())
                .build());
    }

    /**
     * Updates the ref for the repository replica as specified by {@code parameters}, without triggering any hooks.
     *
     * @param node       the node hosting the repository replica to update
     * @param parameters describes the ref to update
     */
    public void updateRefNoHooks(@Nonnull MeshNode node, @Nonnull UpdateRefParameters parameters) {
        ProposedRefUpdate refUpdate = parameters.getUpdate();
        Path repoDir = getRepoDir(node, parameters.getRepository().getId());

        GitUpdateRefBuilderSupport<?> builder;
        if (refUpdate.getType() == RefUpdateType.DELETE) {
            builder = builderFactory.builder().updateRef()
                    .delete(refUpdate.getRefId());
        } else {
            builder = builderFactory.builder().updateRef()
                    .set(refUpdate.getRefId(), refUpdate.getNewHash());
        }
        builder.deref(false)
                .withConfiguration("core.hooksPath", "hooks") // disables the central hooks
                .workDir(repoDir)
                .build()
                .run();
    }

    public void updateRefs(@Nonnull MeshNode node, @Nonnull UpdateRefsParameters parameters) {
        blockingRefStub(node).updateRefs(RpcUpdateRefsRequest.newBuilder()
                .setRepository(parameters.getRepository().getId())
                .addAllRefUpdates(() -> parameters.getUpdates().stream().map(ProposedRefUpdate::toRefUpdate).iterator())
                .setTransactional(parameters.isTransactional())
                .build());
    }

    public RpcRepositoryConsistencySummary verifyConsistency(@Nonnull MeshNode node, @Nonnull String repositoryId) {
        return blockingRepositoryStub(node).verifyConsistency(RpcVerifyRepositoryConsistencyRequest.newBuilder()
                .setRepository(repositoryId)
                .build())
                .getSummary();
    }

    public List<RpcRepositoryIntegrityIssue> verifyIntegrity(@Nonnull MeshNode node, @Nonnull String repositoryId,
                                                             long latestPullRequestId) {
        requireNonNull(node, "node");
        requireNonNull(repositoryId, "repositoryId");

        return blockingRepositoryStub(node).bulkVerifyRepositoryIntegrity(
                RpcBulkVerifyRepositoryIntegrityRequest.newBuilder()
                        .addRepositoryRequests(RpcVerifyRepositoryRequest.newBuilder()
                                .setRepository(repositoryId)
                                .setLatestPullRequestId(latestPullRequestId))
                        .build()).getIntegrityIssuesList();
    }

    private ManagementServiceBlockingStub blockingManagementStub(MeshNode node) {
        return channelRegistry.getChannel(node).newStub(ManagementServiceGrpc::newBlockingStub);
    }

    private RefServiceBlockingStub blockingRefStub(MeshNode node) {
        return channelRegistry.getChannel(node).newStub(RefServiceGrpc::newBlockingStub);
    }

    private RepositoryServiceBlockingStub blockingRepositoryStub(MeshNode node) {
        return channelRegistry.getChannel(node).newStub(RepositoryServiceGrpc::newBlockingStub);
    }

    private void fetch(Path source, Path destination, String... refspecs) {
        executionManager.withContext(ctx -> {
            GitFetchBuilder fetchBuilder = builderFactory.builder().fetch()
                    .workDir(destination)
                    .repository(source);
            for (String refspec : refspecs) {
                fetchBuilder.refspec(refspec);
            }
            //disable central hooks to prevent reference-transaction callbacks
            fetchBuilder.withConfiguration("core.hooksPath", "hooks");
            fetchBuilder.build(new LoggingStdioHandler())
                    .run();
            return null;
        });
    }

    private Path getRepoDir(MeshNode node, Repository repository) {
        if (repository instanceof SharedObjectsRepository) {
            if (node.isLocal()) {
                return repository.getPath();
            }
            return getRepoDir(node, ((SharedObjectsRepository) repository).getPartition(), ((SharedObjectsRepository) repository).getHierarchyId(), "shared-objects");
        } else if (repository instanceof ManagedRepository) {
            return getRepoDir(node, repository.getId());
        }
        throw new IllegalArgumentException("Only ManagedRepository and SharedObjectsRepository are accepted");
    }

    private Path getRepoDir(MeshNode node, ManagedRepository repository) {
        if (node == null || node.isLocal()) {
            return repository.getPath();
        }
        return getRepoDir(node, repository.getPartition(), repository.getHierarchyId(), repository.getLocalId());
    }

    private Path getRepoDir(MeshNode node, String repositoryId) {
        if (node.isLocal()) {
            return repositoryManager.getById(repositoryId).getPath();
        }

        //Only managed repositories are supported on remote nodes
        ManagedRepository.Id id = ManagedRepository.Id.parseId(repositoryId)
                .orElseThrow(
                () -> new IllegalArgumentException(repositoryId + " does not look like a managed repository ID"));
        return getRepoDir(node, id.getPartition(), id.getHierarchyId(), id.getLocalId());
    }

    private Path getRepoDir(MeshNode node, String partition, String hierarchyId, String localId) {
        return clusterManager.getNodeById(node.getId())
                .getRepoDir(partition, hierarchyId, localId);
    }

    private PlumbingServiceStub plumbingStub(MeshNode node) {
        return channelRegistry.getChannel(node).newStub(PlumbingServiceGrpc::newStub);
    }

    private RepositoryServiceStub repositoryStub(MeshNode node) {
        return channelRegistry.getChannel(node).newStub(RepositoryServiceGrpc::newStub);
    }

    private void touchFile(MeshNode node, Repository repository, Path file, Instant timestamp) {
        Path path = getRepoDir(node, repository).resolve(file);
        MoreFiles.mkdir(path.getParent());
        try {
            MoreFiles.touch(path, timestamp);
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
    }

    private static class StringStreamObserver<T> extends AbstractFutureResponseObserver<T, RpcStreamFragment, String> {

        private final ByteArrayOutputStream output = new ByteArrayOutputStream();

        @Override
        public void onNext(RpcStreamFragment fragment) {
            try {
                switch (fragment.getFragmentOneofCase()) {
                    case STDOUT:
                        output.write(fragment.getStdout().toByteArray());
                        break;
                    case RESULT:
                        setResult(output.toString("UTF-8"));
                        break;
                }
            } catch (IOException e) {
                throw new UncheckedIOException(e);
            }
        }
    }
}
