package com.atlassian.bitbucket.mesh;

import com.atlassian.bitbucket.mesh.client.MeshClient;
import com.atlassian.bitbucket.mesh.client.RepairRepositoryParameters;
import com.atlassian.bitbucket.mesh.cluster.WiredClusterNode;
import com.atlassian.bitbucket.mesh.git.GitObjectType;
import com.atlassian.bitbucket.mesh.git.ref.*;
import com.atlassian.bitbucket.mesh.io.MoreFiles;
import com.atlassian.bitbucket.mesh.junit.LogListener;
import com.atlassian.bitbucket.mesh.network.MeshManager;
import com.atlassian.bitbucket.mesh.network.MeshNode;
import com.atlassian.bitbucket.mesh.repair.RepairSummary;
import com.atlassian.bitbucket.mesh.repository.*;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcGetRepositoryDetailsResponse;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcGetRepositoryStateResponse;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcReplicaState;
import com.atlassian.bitbucket.mesh.test.ExecutionContextRule;
import com.atlassian.bitbucket.mesh.test.annotation.DirtiesRepo;
import com.atlassian.bitbucket.mesh.test.annotation.Repo;
import com.atlassian.bitbucket.mesh.test.cluster.MeshClusterManager;
import com.atlassian.bitbucket.mesh.test.cluster.MeshClusterNode;
import com.atlassian.bitbucket.mesh.util.ShaUtils;
import com.atlassian.bitbucket.mesh.util.Version;
import com.google.common.collect.ImmutableMap;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.mutable.MutableObject;
import org.junit.Rule;
import org.junit.Test;

import javax.annotation.Resource;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.List;
import java.util.Map;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Supplier;
import java.util.stream.Collectors;

import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.*;

public class RepairTest extends AbstractWiredTest {

    //Timeout for the repair tests. Most tests should complete under a second, but a much longer timeout is used
    //to account for slow build nodes
    private static final long TIMEOUT_MS = 30000L;

    @Rule
    public final LogListener logListener = new LogListener("com.atlassian.bitbucket.mesh.git.RemoteTransactionLeader");

    @Resource
    @Rule
    public ExecutionContextRule executionContextRule;

    @Repo
    private BitbucketServerRepository bitbucketServerRepository;
    @Resource
    private MeshClusterManager clusterManager;
    @Resource
    private WiredClusterNode localClusterNode;
    @Resource
    private MeshClient meshClient;
    @Resource
    private MeshManager meshManager;
    private MeshNode repairSource;
    private MeshNode repairTarget;
    @Repo
    private ManagedRepository repository;
    @Resource
    private RepositoryHashService repositoryHashService;
    @Resource
    private TestHelper testHelper;

    @DirtiesRepo("repository")
    @Test(timeout = TIMEOUT_MS)
    public void testAutoRepairOnReplicationWhenInconsistent() throws Exception {
        MeshNode node = meshManager.getNodes().stream()
                .filter(MeshNode::isRemote)
                .findAny()
                .orElseThrow(() -> new IllegalStateException("No remote node found"));

        String repositoryId = repository.getId();
        Branch branch = localClusterNode.whileOffline(() -> {
            //With the local node offline, introduce a branch with a new commit on the other nodes
            return testHelper.createBranchWithNewCommit(node, repositoryId, "refs/heads/auto-repair-test");
        });

        //Introduce another, non-conflicting, update. This should trigger this node to detect that it is inconsistent
        //and trigger repair.
        testHelper.createBranchWithNewCommit(node, repositoryId, "refs/heads/auto-repair-test-2");

        Instant deadline = Instant.now().plus(30, ChronoUnit.SECONDS);
        boolean repaired = false;
        while (Instant.now().isBefore(deadline)) {
            Ref ref = testHelper.resolveRef(meshManager.getLocalNode(), repositoryId, branch.getId());
            if (repaired = (ref != null && ref.getLatestCommit().equals(branch.getLatestCommit()))) {
                break;
            }
            Thread.sleep(25L);
        }
        assertTrue("The repository was not repaired!", repaired);
    }

    @DirtiesRepo("repository")
    @Test(timeout = TIMEOUT_MS)
    public void testManagementServiceRepairFromSource() {
        doTestManagementServiceRepair((source, target) -> source);
    }

    @DirtiesRepo("repository")
    @Test(timeout = TIMEOUT_MS)
    public void testManagementServiceRepairFromTarget() {
        doTestManagementServiceRepair((source, target) -> target);
    }

    @Test
    public void testRepairBitbucketServerToManaged() {
        asSource(this::doTestRepairBitbucketServerToManaged);
    }

    @DirtiesRepo("repository")
    @Test(timeout = TIMEOUT_MS)
    public void testStashRefsMappedToDarkRefsOnRepairAsSource() {
        asSource(this::doTestStashRefsMappedToDarkRefsOnRepair);
    }

    @DirtiesRepo("repository")
    @Test(timeout = TIMEOUT_MS)
    public void testStashRefsMappedToDarkRefsOnRepairAsTarget() {
        asTarget(this::doTestStashRefsMappedToDarkRefsOnRepair);
    }

    @DirtiesRepo("repository")
    @Test(timeout = TIMEOUT_MS)
    public void testRepairCanRecoverFromInvalidExistingRef() {
        asTarget(() -> {
            try {
                //Create an invalid ref (a ref that points to a commit that does not exist in the repository)
                MoreFiles.write(repository.getPath().resolve("refs/heads/master"),
                        "d0ba1eb11f58caa8cc6b093aaa8acdb0736fd04c", StandardCharsets.UTF_8);
                //Remove the content-hash to force it to be recalculated and trigger a content hash mismatch
                repositoryHashService.resetContentHash(repository);

                //Repair and content hash calculation uses git ls-remote instead of for-each-ref because it
                //does not fail on invalid refs, at least in git 2.34 and higher. For 2.33 and lower, we expect
                //the repair to  fail
                String repositoryId = repository.getId();
                if (isGitAtLeast(new Version(2, 34))) {
                    //Try to repair the repository
                    repair(repositoryId);

                    //Verify that the invalid master branch has been repaired to the expected value
                    Ref master = testHelper.resolveRef(repairTarget, repositoryId, "refs/heads/master");
                    assertNotNull(master);
                    assertEquals("0a943a29376f2336b78312d99e65da17048951db", master.getLatestCommit());
                } else {
                    assertThrows(RuntimeException.class, () -> repair(repositoryId));
                    //Make the repository valid again to avoid problems during test teardown
                    MoreFiles.deleteQuietly(repository.getPath().resolve("refs/heads/master"));
                }
            } catch (IOException e) {
                throw new UncheckedIOException(e);
            }
        });
    }

    @DirtiesRepo("repository")
    @Test(timeout = TIMEOUT_MS)
    public void testRepairCanRepairReadOnlyRepository() {
        asTarget(() -> {
            MutableObject<Branch> createdBranch = new MutableObject<>();

            String repositoryId = repository.getId();
            withRepairTargetOffline(() -> {
                ProposedRefUpdate refUpdate = ProposedRefUpdate.of("refs/heads/branch_mod_merge",
                        "6053a1eaa1c009dd11092d09a72f3c41af1b59ad", "9c05f43f859375e392d90d23a13717c16d0fdcda");
                testHelper.updateRef(repairSource, new UpdateRefParameters.Builder(repository, refUpdate).build());
                createdBranch.setValue(
                        testHelper.createBranchWithNewCommit(repairSource, repositoryId, "refs/heads/new-branch"));
            });
            testHelper.setReadOnly(repairTarget, repositoryId, true);

            try {
                repair(repositoryId);

                Ref resolved = testHelper.resolveRef(repairTarget, repositoryId, createdBranch.getValue().getId());
                assertNotNull(resolved);
                assertEquals(createdBranch.getValue().getLatestCommit(), resolved.getLatestCommit());
            } finally {
                testHelper.setReadOnly(repairTarget, repositoryId, false);
            }
        });
    }

    @DirtiesRepo("repository")
    @Test(timeout = TIMEOUT_MS)
    public void testRepairDefaultBranchMismatchAsSource() {
        asSource(this::doTestRepairDefaultBranchMismatch);
    }

    @DirtiesRepo("repository")
    @Test(timeout = TIMEOUT_MS)
    public void testRepairDefaultBranchMismatchAsTarget() {
        asTarget(this::doTestRepairDefaultBranchMismatch);
    }

    @DirtiesRepo("repository")
    @Test(timeout = TIMEOUT_MS)
    public void testRepairHandlesInvalidCommitIdInDarkRefAsSource() {
        asSource(this::doTestRepairHandlesInvalidCommitIdInDarkRef);
    }

    @DirtiesRepo("repository")
    @Test(timeout = TIMEOUT_MS)
    public void testRepairHandlesInvalidCommitIdInDarkRefAsTarget() {
        asTarget(this::doTestRepairHandlesInvalidCommitIdInDarkRef);
    }

    @DirtiesRepo("repository")
    @Test(timeout = TIMEOUT_MS)
    public void testRepairHandlesInvalidCommitIdInReflogAsSource() {
        asSource(this::doTestRepairHandlesInvalidCommitIdInReflog);
    }

    @DirtiesRepo("repository")
    @Test(timeout = TIMEOUT_MS)
    public void testRepairHandlesInvalidCommitIdInReflogAsTarget() {
        asTarget(this::doTestRepairHandlesInvalidCommitIdInReflog);
    }

    @Test
    public void testRepairManagedToBitbucketServer() {
        asTarget(this::doTestRepairManagedToBitbucketServer);
    }

    @Test(timeout = TIMEOUT_MS)
    public void testRepairMissingEmptyRepositoryAsSource() {
        asSource(this::doTestRepairMissingEmptyRepository);
    }

    @Test(timeout = TIMEOUT_MS)
    public void testRepairMissingEmptyRepositoryAsTarget() {
        asTarget(this::doTestRepairMissingEmptyRepository);
    }

    @Test(timeout = TIMEOUT_MS)
    public void testRepairMissingRepositoryAsSource() {
        asSource(this::doTestRepairMissingRepository);
    }

    @Test(timeout = TIMEOUT_MS)
    public void testRepairMissingRepositoryAsTarget() {
        asTarget(this::doTestRepairMissingRepository);
    }

    @DirtiesRepo("repository")
    @Test(timeout = TIMEOUT_MS)
    public void testRepairRefMismatchIntroducingNewObjectsAsSource() {
        asSource(this::doTestRepairRefMismatchIntroducingNewObjects);
    }

    @DirtiesRepo("repository")
    @Test(timeout = TIMEOUT_MS)
    public void testRepairRefMismatchIntroducingNewObjectsAsTarget() {
        asTarget(this::doTestRepairRefMismatchIntroducingNewObjects);
    }

    @DirtiesRepo("repository")
    @Test(timeout = TIMEOUT_MS)
    public void testRepairRefMismatchIntroducingNewObjectsReferencedByReflogOnlyAsSource() {
        asSource(this::doTestRepairRefMismatchIntroducingNewObjectsReferencedByReflogsOnly);
    }

    @DirtiesRepo("repository")
    @Test(timeout = TIMEOUT_MS)
    public void testRepairRefMismatchIntroducingNewObjectsReferencedByReflogOnlyAsTarget() {
        asTarget(this::doTestRepairRefMismatchIntroducingNewObjectsReferencedByReflogsOnly);
    }

    @DirtiesRepo("repository")
    @Test(timeout = TIMEOUT_MS)
    public void testStashReflogsMappedToDarkReflogsOnRepairAsSource() {
        asSource(this::doTestStashReflogsMappedToDarkReflogsOnRepair);
    }

    @DirtiesRepo("repository")
    @Test(timeout = TIMEOUT_MS)
    public void testStashReflogsMappedToDarkReflogsOnRepairAsTarget() {
        asTarget(this::doTestStashReflogsMappedToDarkReflogsOnRepair);
    }

    @DirtiesRepo("repository")
    @Test(timeout = TIMEOUT_MS)
    public void testDuplicateStashAndDarkReflogsMappedToDarkReflogsOnRepair() {
        asTarget(() -> {
            String repositoryId = repository.getId();
            String reflog = ShaUtils.NULL_SHA1 + " 9c05f43f859375e392d90d23a13717c16d0fdcda " +
                    "Manish Goyal <mgoyal2@atlassian.com> 1488149733 +0000\n";
            withRepairTargetOffline(() -> {
                ProposedRefUpdate refUpdate = ProposedRefUpdate.of("refs/heads/branch_mod_merge",
                        "6053a1eaa1c009dd11092d09a72f3c41af1b59ad",
                        "9c05f43f859375e392d90d23a13717c16d0fdcda");
                testHelper.updateRef(repairSource, new UpdateRefParameters.Builder(repository, refUpdate).build());
                testHelper.createLooseRef(repairSource, repository, "stash-refs/pull-requests/5634/from",
                        "9c05f43f859375e392d90d23a13717c16d0fdcda");
                testHelper.createLooseRef(repairSource, repository, "dark-refs/pull-requests/5634/from",
                        "9c05f43f859375e392d90d23a13717c16d0fdcda");
                testHelper.setReflog(repairSource, repository, "stash-refs/pull-requests/5634/from", reflog);
                testHelper.setReflog(repairSource, repository, "dark-refs/pull-requests/5634/from", reflog);
            });

            try {
                repair(repositoryId);

                assertNull(testHelper.getLooseRef(repairTarget, repository, "stash-refs/pull-requests/5634/from"));
                String looseRef = testHelper.getLooseRef(repairTarget, repository,
                        "dark-refs/pull-requests/5634/from");
                assertEquals("9c05f43f859375e392d90d23a13717c16d0fdcda", looseRef);


                assertNull(testHelper.getReflog(repairTarget, repository, "stash-refs/pull-requests/5634/from"));
                assertEquals(reflog,
                        testHelper.getReflog(repairTarget, repository, "dark-refs/pull-requests/5634/from"));
            } finally {
                testHelper.setReadOnly(repairTarget, repositoryId, false);
            }
        });
    }

    @DirtiesRepo("repository")
    @Test(timeout = TIMEOUT_MS)
    public void testRepairRefMismatchNoNewObjectsAsSource() {
        asSource(this::doTestRepairRefMismatchNoNewObjects);
    }

    @DirtiesRepo("repository")
    @Test(timeout = TIMEOUT_MS)
    public void testRepairRefMismatchNoNewObjectsAsTarget() {
        asTarget(this::doTestRepairRefMismatchNoNewObjects);
    }

    @DirtiesRepo("repository")
    @Test(timeout = TIMEOUT_MS)
    public void testRepairRefreshesContentHashOnSourceIfHashesMismatchButNoRefsDiffer() {
        ProposedRefUpdate refUpdate = ProposedRefUpdate.of("refs/heads/branch_mod_merge",
                "6053a1eaa1c009dd11092d09a72f3c41af1b59ad", "9c05f43f859375e392d90d23a13717c16d0fdcda");
        UpdateRefParameters parameters = new UpdateRefParameters.Builder(repository, refUpdate).build();
        RpcGetRepositoryStateResponse beforeState = testHelper.getReplicaState(
                meshManager.getLocalNode(), repository.getId());
        asSource(() -> {
            withRepairSourceOffline(() -> {
                //Update the ref on all nodes but repairSource
                testHelper.updateRef(repairTarget, parameters);
                //Update the ref on repairSource without triggering the hooks, thereby leaving the content-hash
                //at its old value. When repairing, a content-hash mismatch should be detected without any out-of-sync
                //refs. This in turn should trigger a refresh of the content hashes after which the repositories should
                //be back in sync.
                testHelper.updateRefNoHooks(repairSource, parameters);
            });

            String repositoryId = repository.getId();
            repair(repositoryId);

            //Retrieve the replica state for source and target and validate that their hashes match
            RpcGetRepositoryStateResponse sourceState = testHelper.getReplicaState(repairSource, repositoryId);
            RpcGetRepositoryStateResponse targetState = testHelper.getReplicaState(repairTarget, repositoryId);

            //Double check that the content hash has actually changed as a result of the ref updates & repair
            assertNotEquals("Content hash has not changed", beforeState.getContentHash(), sourceState.getContentHash());

            assertEquals("Source is not consistent", RpcReplicaState.REPLICA_STATE_CONSISTENT, sourceState.getState());
            assertEquals("Target is not consistent", RpcReplicaState.REPLICA_STATE_CONSISTENT, targetState.getState());
            assertEquals("Content hashes do not match", sourceState.getContentHash(), targetState.getContentHash());
            assertEquals("Metadata hashes do not match", sourceState.getMetadataHash(), targetState.getMetadataHash());
        });
    }

    @DirtiesRepo("repository")
    @Test(timeout = TIMEOUT_MS)
    public void testRepairRefreshesContentHashOnTargetIfHashesMismatchButNoRefsDiffer() {
        ProposedRefUpdate refUpdate = ProposedRefUpdate.of("refs/heads/branch_mod_merge",
                "6053a1eaa1c009dd11092d09a72f3c41af1b59ad", "9c05f43f859375e392d90d23a13717c16d0fdcda");
        UpdateRefParameters parameters = new UpdateRefParameters.Builder(repository, refUpdate).build();
        RpcGetRepositoryStateResponse beforeState = testHelper.getReplicaState(
                meshManager.getLocalNode(), repository.getId());
        asSource(() -> {
            withRepairTargetOffline(() -> {
                //Update the ref on all nodes but repairTarget
                testHelper.updateRef(repairSource, parameters);
                //Update the ref on repairTarget without triggering the hooks, thereby leaving the content-hash
                //at its old value. When repairing, a content-hash mismatch should be detected without any out-of-sync
                //refs. This in turn should trigger a refresh of the content hashes after which the repositories should
                //be back in sync.
                testHelper.updateRefNoHooks(repairTarget, parameters);
            });

            String repositoryId = repository.getId();
            repair(repositoryId);

            //Retrieve the replica state for source and target and validate that their hashes match
            RpcGetRepositoryStateResponse sourceState = testHelper.getReplicaState(repairSource, repositoryId);
            RpcGetRepositoryStateResponse targetState = testHelper.getReplicaState(repairTarget, repositoryId);

            //Double check that the content hash has actually changed as a result of the ref updates & repair
            assertNotEquals("Content hash has not changed", beforeState.getContentHash(), sourceState.getContentHash());

            assertEquals("Source is not consistent", RpcReplicaState.REPLICA_STATE_CONSISTENT, sourceState.getState());
            assertEquals("Target is not consistent", RpcReplicaState.REPLICA_STATE_CONSISTENT, targetState.getState());
            assertEquals("Content hashes do not match", sourceState.getContentHash(), targetState.getContentHash());
            assertEquals("Metadata hashes do not match", sourceState.getMetadataHash(), targetState.getMetadataHash());
        });
    }

    @DirtiesRepo("repository")
    @Test(timeout = TIMEOUT_MS)
    public void testRepairTranscodeMismatchAsSource() {
        asSource(this::doTestRepairTranscodeMismatch);
    }

    @DirtiesRepo("repository")
    @Test(timeout = TIMEOUT_MS)
    public void testRepairTranscodeMismatchAsTarget() {
        asTarget(this::doTestRepairTranscodeMismatch);
    }

    private static String generateLocalId() {
        return Long.toHexString(System.nanoTime());
    }

    private void asSource(Runnable action) {
        repairSource = meshManager.getLocalNode();
        repairTarget = getRemoteNode();

        action.run();
    }

    private void asTarget(Runnable action) {
        repairSource = getRemoteNode();
        repairTarget = meshManager.getLocalNode();

        action.run();
    }

    private void doTestManagementServiceRepair(BiFunction<MeshNode, MeshNode, MeshNode> nodePicker) {
        //Select both source and target to be remote
        List<MeshNode> remoteNodes = meshManager.getNodes().stream()
                .filter(MeshNode::isRemote)
                .collect(Collectors.toList());

        assertTrue(remoteNodes.size() >= 2);
        repairSource = remoteNodes.get(0);
        repairTarget = remoteNodes.get(1);

        ProposedRefUpdate refUpdate = ProposedRefUpdate.of("refs/heads/branch_mod_merge",
                "6053a1eaa1c009dd11092d09a72f3c41af1b59ad", "9c05f43f859375e392d90d23a13717c16d0fdcda");
        UpdateRefParameters parameters = new UpdateRefParameters.Builder(repository, refUpdate).build();
        RpcGetRepositoryStateResponse beforeState = testHelper.getReplicaState(repairTarget, repository.getId());
        withNodeOffline(repairTarget, () -> {
            //Update a ref on the repair source that is not synced to repair target
            testHelper.updateRef(repairSource, parameters);
        });

        MeshNode node = nodePicker.apply(repairSource, repairTarget);
        //Instruct the selected node to repair the target from the source
        ReplicaState state = testHelper.repair(node, repairSource, repairTarget, repository.getId());

        //Validate that target has been repaired
        assertEquals(ReplicaState.CONSISTENT, state);
        RpcGetRepositoryStateResponse afterState = testHelper.getReplicaState(repairTarget, repository.getId());
        assertEquals(RpcReplicaState.REPLICA_STATE_CONSISTENT, afterState.getState());
        assertNotEquals(beforeState.getContentHash(), afterState.getContentHash());
        Ref resolved = testHelper.resolveRef(repairTarget, repository.getId(), "refs/heads/branch_mod_merge");
        assertNotNull(resolved);
        assertEquals("9c05f43f859375e392d90d23a13717c16d0fdcda", resolved.getLatestCommit());
    }

    private void doTestRepairBitbucketServerToManaged() {
        String managedRepositoryId = ManagedRepository.formatId(0, bitbucketServerRepository.getHierarchyId(),
                generateLocalId());

        doTestRepairToDifferentType(bitbucketServerRepository.getId(), managedRepositoryId);
    }

    private void doTestRepairDefaultBranchMismatch() {
        String defaultBranch = "refs/heads/basic_branching";
        String repositoryId = repository.getId();

        //Update the repair source while the target is unavailable
        withRepairTargetOffline(() -> testHelper.setDefaultBranch(repairSource, repositoryId, defaultBranch));

        //Verify that the repair target is indeed out of sync
        assertNotEquals(defaultBranch, testHelper.getDefaultBranch(repairTarget, repository.getId()));

        repair(repositoryId);

        //Verify that the node has been repaired
        assertEquals("Default branch does not match", defaultBranch,
                testHelper.getDefaultBranch(repairTarget, repository.getId()));
    }

    private void doTestRepairHandlesInvalidCommitIdInDarkRef() {
        MutableObject<Branch> newBranch = new MutableObject<>();
        String repositoryId = repository.getId();
        withRepairTargetOffline(() -> {
            //Create a new branch to make sure there's something to sync
            newBranch.setValue(testHelper.createBranchWithNewCommit(
                    repairSource, repositoryId, "refs/heads/new-branch"));
            //Create a dark ref that references an object that does not exist in the source. Git handles this fine,
            //so we want to validate that repair does not break when recreating the dark ref on the target
            testHelper.createLooseRef(repairSource, repository, "dark-refs/pull-requests/1234/from",
                    "e04488950e0448cc5e1def735d55a299f8159369");
        });

        repair(repositoryId);

        //Verify that the reflog has been synced, and that the refs have been synced correctly
        assertEquals(newBranch.getValue(), testHelper.resolveRef(repairTarget, repositoryId, "refs/heads/new-branch"));
        assertEquals("e04488950e0448cc5e1def735d55a299f8159369",
                testHelper.getLooseRef(repairTarget, repository, "dark-refs/pull-requests/1234/from"));
    }

    private void doTestRepairHandlesInvalidCommitIdInReflog() {
        MutableObject<Branch> newBranch = new MutableObject<>();
        String reflog = ShaUtils.NULL_SHA1 + " e04488950e0448cc5e1def735d55a299f8159369 " +
                "John Doe <jdoe@email.com> 1488149733 +0000\n";
        String repositoryId = repository.getId();

        withRepairTargetOffline(() -> {
            //Create a new branch to make sure there's something to sync
            newBranch.setValue(testHelper.createBranchWithNewCommit(
                    repairSource, repositoryId, "refs/heads/new-branch"));
            //Create a reflog that references an object that does not exist in the source. Git handles this fine,
            //so we want to validate that repair does not break when it cannot include these objects in the
            //missing objects pack.
            testHelper.setReflog(repairSource, repository, "refs/heads/stale-branch", reflog);
        });

        repair(repositoryId);

        //Verify that the reflog has been synced, and that the refs have been synced correctly
        assertEquals(reflog, testHelper.getReflog(repairTarget, repository, "refs/heads/stale-branch"));
        assertEquals(newBranch.getValue(), testHelper.resolveRef(repairTarget, repositoryId, "refs/heads/new-branch"));
    }

    private void doTestRepairManagedToBitbucketServer() {
        String bitbucketServerRepositoryId = BitbucketServerRepository.formatId(
                repository.getHierarchyId(), generateLocalId());

        doTestRepairToDifferentType(repository.getId(), bitbucketServerRepositoryId);
    }

    private void doTestRepairMissingEmptyRepository() {
        doTestRepairMissingRepository(null, null);
    }

    private void doTestRepairMissingRepository() {
        MutableObject<String> branchCommitId = new MutableObject<>();
        MutableObject<String> tagCommitId = new MutableObject<>();
        doTestRepairMissingRepository(
                repositoryId -> {
                    //Create a branch and a tag in the newly created empty repository
                    branchCommitId.setValue(testHelper.createBranchWithNewCommit(
                            repairSource, repositoryId, "refs/heads/this-is-a-branch").getLatestCommit());
                    tagCommitId.setValue(testHelper.createBranchWithNewCommit(
                            repairSource, repositoryId, "refs/tags/this-is-a-tag").getLatestCommit());
                },
                repositoryId -> {
                    //Verify that the refs have been synced
                    Ref resolved = testHelper.resolveRef(repairTarget, repositoryId, "refs/heads/this-is-a-branch");
                    assertThat(resolved, instanceOf(Branch.class));
                    assertEquals(branchCommitId.getValue(), resolved.getLatestCommit());
                    resolved = testHelper.resolveRef(repairTarget, repositoryId, "refs/tags/this-is-a-tag");
                    assertThat(resolved, instanceOf(Tag.class));
                    assertEquals(tagCommitId.getValue(), resolved.getLatestCommit());
                });
    }

    private void doTestRepairMissingRepository(Consumer<String> repositoryInitializer,
                                               Consumer<String> additionalVerifier) {
        String localId = generateLocalId();
        String hierarchyId = StringUtils.leftPad(localId, 20, '0');
        String newRepoId = ManagedRepository.formatId(9, hierarchyId, localId);
        String defaultBranch = "refs/heads/magic";
        Map<String, String> metadata = ImmutableMap.of("foo", "bar");

        withRepairTargetOffline(() -> {
            //Create a repository and enable transcoding
            testHelper.createRepository(repairSource, new CreateRepositoryParameters.Builder(newRepoId)
                    .metadata(metadata)
                    .defaultBranch(defaultBranch)
                    .build());

            testHelper.setTranscodeEnabled(repairSource, newRepoId, true);

            if (repositoryInitializer != null) {
                repositoryInitializer.accept(newRepoId);
            }
        });

        repair(newRepoId);

        //Verify that the repository replica on the repair target has been repaired
        RpcGetRepositoryDetailsResponse details = testHelper.getDetails(repairTarget, newRepoId);
        assertEquals("Default branch does not match", defaultBranch, details.getDefaultBranch().toStringUtf8());
        assertEquals("Metadata does not match", metadata, details.getMetadataMap());
        assertTrue("Transcode is not enabled", details.getTranscodeEnabled());

        if (additionalVerifier != null) {
            additionalVerifier.accept(newRepoId);
        }

        testHelper.deleteRepository(repairSource, newRepoId);
    }

    private void doTestRepairRefMismatchIntroducingNewObjects() {
        MutableObject<Branch> createdBranch = new MutableObject<>();
        String repositoryId = repository.getId();

        withRepairTargetOffline(() -> {
            ProposedRefUpdate refUpdate = ProposedRefUpdate.of("refs/heads/branch_mod_merge",
                    "6053a1eaa1c009dd11092d09a72f3c41af1b59ad", "9c05f43f859375e392d90d23a13717c16d0fdcda");
            testHelper.updateRef(repairSource, new UpdateRefParameters.Builder(repository, refUpdate).build());
            createdBranch.setValue(
                    testHelper.createBranchWithNewCommit(repairSource, repositoryId, "refs/heads/new-branch"));
        });

        repair(repositoryId);

        Ref resolved = testHelper.resolveRef(repairTarget, repositoryId, "refs/heads/branch_mod_merge");
        assertNotNull(resolved);
        assertEquals("9c05f43f859375e392d90d23a13717c16d0fdcda", resolved.getLatestCommit());

        resolved = testHelper.resolveRef(repairTarget, repositoryId, createdBranch.getValue().getId());
        assertNotNull(resolved);
        assertEquals(createdBranch.getValue().getLatestCommit(), resolved.getLatestCommit());
    }

    private void doTestRepairRefMismatchIntroducingNewObjectsReferencedByReflogsOnly() {
        MutableObject<Branch> createdBranch = new MutableObject<>();
        MutableObject<String> darkCommitId = new MutableObject<>();
        MutableObject<String> reflog = new MutableObject<>();
        String repositoryId = repository.getId();

        withRepairTargetOffline(() -> {
            ProposedRefUpdate refUpdate = ProposedRefUpdate.of("refs/heads/branch_mod_merge",
                    "6053a1eaa1c009dd11092d09a72f3c41af1b59ad", "9c05f43f859375e392d90d23a13717c16d0fdcda");
            testHelper.updateRef(repairSource, new UpdateRefParameters.Builder(repository, refUpdate).build());
            createdBranch.setValue(
                    testHelper.createBranchWithNewCommit(repairSource, repositoryId, "refs/heads/new-branch"));

            //Create a commit in the repair source's replica
            darkCommitId.setValue(testHelper.createCommit(repairSource, repositoryId));

            //Protect the newly created commit with a reflog under a different name so it won't be deleted
            //when we delete the ref itself.
            Supplier<String> userDetails = () -> " Michael Heemskerk <mheemskerk@atlassian.com> " +
                    System.currentTimeMillis() / 1000 + " +0000\n";
            reflog.setValue(ShaUtils.NULL_SHA1 + " " + darkCommitId.getValue() + userDetails.get() +
                    darkCommitId.getValue() + " " + ShaUtils.NULL_SHA1 + userDetails.get());
            testHelper.setReflog(repairSource, repository, "dark-refs/heads/tmp-branch", reflog.getValue());
        });

        repair(repositoryId);

        //Validate that the added ref is synced
        Ref resolved = testHelper.resolveRef(repairTarget, repositoryId, "refs/heads/branch_mod_merge");
        assertNotNull(resolved);
        assertEquals("9c05f43f859375e392d90d23a13717c16d0fdcda", resolved.getLatestCommit());

        resolved = testHelper.resolveRef(repairTarget, repositoryId, createdBranch.getValue().getId());
        assertNotNull(resolved);
        assertEquals(createdBranch.getValue().getLatestCommit(), resolved.getLatestCommit());

        //Validate that the reflog is synced
        String syncedReflog = testHelper.getReflog(repairTarget, repository, "dark-refs/heads/tmp-branch");
        assertNotNull("Reflog was not synced", syncedReflog);
        assertEquals("Reflog was not synced correctly", reflog.getValue(), syncedReflog);

        //Validate that the synced reflog is valid
        assertEquals("Commit that is only referenced by reflog was not synced",
                GitObjectType.COMMIT, testHelper.getObjectType(repairTarget, repository, darkCommitId.getValue()));
    }

    private void doTestRepairRefMismatchNoNewObjects() {
        String repositoryId = repository.getId();

        withRepairTargetOffline(() -> {
            ProposedRefUpdate refUpdate = ProposedRefUpdate.of("refs/heads/branch_mod_merge",
                    "6053a1eaa1c009dd11092d09a72f3c41af1b59ad", "9c05f43f859375e392d90d23a13717c16d0fdcda");
            testHelper.updateRef(repairSource, new UpdateRefParameters.Builder(repository, refUpdate).build());
        });

        repair(repositoryId);

        Ref resolved = testHelper.resolveRef(repairTarget, repositoryId, "refs/heads/branch_mod_merge");
        assertNotNull(resolved);
        assertEquals("9c05f43f859375e392d90d23a13717c16d0fdcda", resolved.getLatestCommit());
    }

    private void doTestRepairToDifferentType(String sourceRepositoryId, String targetRepositoryId) {
        try {
            repair(sourceRepositoryId, targetRepositoryId);

            RpcGetRepositoryStateResponse sourceState = testHelper.getReplicaState(repairSource, sourceRepositoryId);
            RpcGetRepositoryStateResponse targetState = testHelper.getReplicaState(repairTarget, targetRepositoryId);

            assertEquals("Source is not consistent", RpcReplicaState.REPLICA_STATE_CONSISTENT, sourceState.getState());
            assertEquals("Target is not consistent", RpcReplicaState.REPLICA_STATE_CONSISTENT, targetState.getState());
            assertEquals("Content hashes do not match", sourceState.getContentHash(), targetState.getContentHash());
            assertEquals("Metadata hashes do not match", sourceState.getMetadataHash(), targetState.getMetadataHash());
        } finally {
            testHelper.deleteRepositoryWithoutCoordination(repairTarget, targetRepositoryId);
        }
    }

    private void doTestRepairTranscodeMismatch() {
        String repositoryId = repository.getId();

        //Update the repair source while the target is unavailable
        withRepairTargetOffline(() -> testHelper.setTranscodeEnabled(repairSource, repositoryId, true));

        //Verify that the repair target is indeed out of sync
        assertFalse(testHelper.isTranscodeEnabled(repairTarget, repositoryId));

        repair(repositoryId);

        //Verify that the node has been repaired
        assertTrue("Transcode is not enabled", testHelper.isTranscodeEnabled(repairTarget, repositoryId));
    }

    private void doTestStashReflogsMappedToDarkReflogsOnRepair() {
        String repositoryId = repository.getId();
        String reflog = ShaUtils.NULL_SHA1 + " 9c05f43f859375e392d90d23a13717c16d0fdcda " +
                "Manish Goyal <mgoyal2@atlassian.com> 1488149733 +0000\n";
        withRepairTargetOffline(() -> {
            ProposedRefUpdate refUpdate = ProposedRefUpdate.of("refs/heads/branch_mod_merge",
                    "6053a1eaa1c009dd11092d09a72f3c41af1b59ad",
                    "9c05f43f859375e392d90d23a13717c16d0fdcda");
            testHelper.updateRef(repairSource, new UpdateRefParameters.Builder(repository, refUpdate).build());
            testHelper.createLooseRef(repairSource, repository, "stash-refs/pull-requests/5634/from",
                    "9c05f43f859375e392d90d23a13717c16d0fdcda");
            testHelper.setReflog(repairSource, repository, "stash-refs/pull-requests/5634/from", reflog);
            testHelper.setReflog(repairSource, repository, "dark-refs/heads/tmp-branch", reflog);
        });

        try {
            repair(repositoryId);

            assertNull(testHelper.getReflog(repairTarget, repository, "stash-refs/pull-requests/5634/from"));
            assertEquals(reflog,
                    testHelper.getReflog(repairTarget, repository, "dark-refs/pull-requests/5634/from"));
            assertEquals(reflog,
                    testHelper.getReflog(repairTarget, repository, "dark-refs/heads/tmp-branch"));
        } finally {
            testHelper.setReadOnly(repairTarget, repositoryId, false);
        }
    }

    private void doTestStashRefsMappedToDarkRefsOnRepair() {
        String repositoryId = repository.getId();
        withRepairTargetOffline(() -> {
            ProposedRefUpdate refUpdate = ProposedRefUpdate.of("refs/heads/branch_mod_merge",
                    "6053a1eaa1c009dd11092d09a72f3c41af1b59ad", "9c05f43f859375e392d90d23a13717c16d0fdcda");
            testHelper.updateRef(repairSource, new UpdateRefParameters.Builder(repository, refUpdate).build());
            testHelper.createLooseRef(repairSource, repository, "stash-refs/pull-requests/5634/from",
                    "9c05f43f859375e392d90d23a13717c16d0fdcda");
        });

        try {
            repair(repositoryId);

            assertNull(testHelper.getLooseRef(repairTarget, repository,
                    "stash-refs/pull-requests/5634/from"));
            String looseRef = testHelper.getLooseRef(repairTarget, repository,
                    "dark-refs/pull-requests/5634/from");
            assertEquals("9c05f43f859375e392d90d23a13717c16d0fdcda", looseRef);
        } finally {
            testHelper.setReadOnly(repairTarget, repositoryId, false);
        }
    }

    private MeshNode getRemoteNode() {
        return meshManager.getNodes().stream()
                .filter(MeshNode::isRemote)
                .findFirst()
                .orElseThrow(() -> new IllegalStateException("No remote replica found"));
    }

    private void repair(String repositoryId) {
        repair(repositoryId, repositoryId);
    }

    private void repair(String sourceRepositoryId, String targetRepositoryId) {
        RepairRepositoryParameters parameters = new RepairRepositoryParameters.Builder(targetRepositoryId)
                .sourceRepository(sourceRepositoryId)
                .timeout(60L)
                .build();
        if (repairSource.isLocal()) {
            meshClient.repair(repairTarget, parameters);
        } else {
            RepairSummary summary = meshClient.repairMe(repairSource, parameters);
            assertEquals("Replica of " + sourceRepositoryId + " is not repaired",
                    ReplicaState.CONSISTENT, summary.getTargetState());
        }
    }

    private void withNodeOffline(MeshNode node, Runnable action) {
        MeshClusterNode clusterNode = node.isLocal() ?
                localClusterNode :
                clusterManager.getNodeById(repairTarget.getId());

        try {
            clusterNode.whileOffline(() -> {
                action.run();
                return null;
            });
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    private void withRepairSourceOffline(Runnable action) {
        withNodeOffline(repairSource, action);
    }

    private void withRepairTargetOffline(Runnable action) {
        withNodeOffline(repairTarget, action);
    }
}
