package com.atlassian.bitbucket.mesh;

import com.atlassian.bitbucket.mesh.cluster.WiredClusterNode;
import com.atlassian.bitbucket.mesh.git.ref.ProposedRefUpdate;
import com.atlassian.bitbucket.mesh.git.ref.UpdateRefParameters;
import com.atlassian.bitbucket.mesh.network.MeshManager;
import com.atlassian.bitbucket.mesh.network.MeshNode;
import com.atlassian.bitbucket.mesh.repository.ManagedRepository;
import com.atlassian.bitbucket.mesh.repository.ReplicaState;
import com.atlassian.bitbucket.mesh.repository.RepositoryManager;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcReplicaState;
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.google.common.collect.ImmutableList;
import io.grpc.StatusRuntimeException;
import org.junit.Before;
import org.junit.Test;

import javax.annotation.Resource;
import java.time.Duration;
import java.util.Collection;
import java.util.Deque;
import java.util.LinkedList;
import java.util.Objects;
import java.util.stream.Stream;

import static com.atlassian.bitbucket.mesh.repository.ReplicaState.CONSISTENT;
import static com.atlassian.bitbucket.mesh.rpc.v1.RpcReplicaState.REPLICA_STATE_CONSISTENT;
import static com.atlassian.bitbucket.mesh.rpc.v1.RpcReplicaState.REPLICA_STATE_INCONSISTENT;
import static com.google.common.base.Preconditions.checkArgument;
import static java.util.stream.Collectors.toList;
import static org.hamcrest.CoreMatchers.containsString;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.contains;
import static org.hamcrest.Matchers.hasSize;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThrows;

public class ContentHashInconsistencyTest extends AbstractWiredTest {

    private Deque<String> availableCommitHashes;
    @Resource
    private MeshClusterManager clusterManager;
    @Resource
    private WiredClusterNode localClusterNode;
    @Resource
    private MeshManager meshManager;
    private MeshNode remainingNode;
    private MeshNode repairSource;
    private MeshNode repairTarget;
    @Repo
    private ManagedRepository repository;
    @Resource
    private RepositoryManager repositoryManager;
    @Resource
    private TestHelper testHelper;

    @Before
    public void setup() {
        availableCommitHashes = new LinkedList<>(ImmutableList.of(
                "9c05f43f859375e392d90d23a13717c16d0fdcda",
                "e0a3f6d8c804c375f49d7609328111e8dc75a4bc",
                "8fb8538799b2481750eac52716931d895d029b7c"));

        assertThat(meshManager.getNodes(), hasSize(3));
    }

    @DirtiesRepo
    @Test
    public void testInconsistentContentHashCausesInconsistency_WithoutForcedContentHashRecalculation() {
        setForcedContentHashRecalculation(false);

        createInconsistentContentHashThroughRepair();

        addNewBranch("refs/heads/new-branch", "0a943a29376f2336b78312d99e65da17048951db");

        Stream.of(repairSource, repairTarget).forEach(node -> assertNodeState(node, REPLICA_STATE_CONSISTENT));
        //Because we didn't forcefully recalculate the content hashes and because the remaining repository had a different
        //content hash than the other two repositories, it's now marked as inconsistent.
        assertNodeState(remainingNode, REPLICA_STATE_INCONSISTENT);

        //However, it should be scheduled for repair and eventually become consistent again.
        awaitUntil(() -> testHelper.getReplicaState(remainingNode, repository.getId()).getState(),
                is(REPLICA_STATE_CONSISTENT),
                Duration.ofSeconds(20L), Duration.ofSeconds(1L));
    }

    @DirtiesRepo
    @Test
    public void testInconsistentContentHashCausesNoInconsistency_WithForcedContentHashRecalculation() {
        setForcedContentHashRecalculation(true);

        createInconsistentContentHashThroughRepair();

        addNewBranch("refs/heads/new-branch", "0a943a29376f2336b78312d99e65da17048951db");

        //With forced hash recalculation, it remains consistent.
        Stream.of(repairSource, repairTarget, remainingNode)
                .forEach(node -> assertNodeState(node, REPLICA_STATE_CONSISTENT));
    }

    @DirtiesRepo
    @Test
    public void testInconsistentRepositoryRefsCauseTransactionRollback_WithForcedRecalculation() {
        Collection<MeshNode> nodes = meshManager.getNodes();

        setForcedContentHashRecalculation(true);

        //Make nodes all inconsistent with each other
        updateRefWithoutHooksToRandomHash(nodes);

        //Run a transaction
        StatusRuntimeException ex = assertThrows(StatusRuntimeException.class, () ->
                addNewBranch("refs/heads/new-branch", "0a943a29376f2336b78312d99e65da17048951db"));
        assertThat(ex.getMessage(), containsString("fatal: ref updates aborted by hook"));

        //While the transaction rolled back, some nodes were marked as inconsistent, these should eventually become
        //consistent again through repair
        awaitUntil(() ->
                        nodes.stream()
                                .map(node -> testHelper.getReplicaState(node, repository.getId()).getState())
                                .collect(toList()),
                contains(REPLICA_STATE_CONSISTENT, REPLICA_STATE_CONSISTENT, REPLICA_STATE_CONSISTENT),
                Duration.ofSeconds(20L), Duration.ofSeconds(1L));
    }

    @DirtiesRepo
    @Test
    public void testInconsistentRepositoryRefsCommitsCleanly_WithoutForcedRecalculation() {
        Collection<MeshNode> nodes = meshManager.getNodes();

        setForcedContentHashRecalculation(false);

        updateRefWithoutHooksToRandomHash(nodes);

        //Run a transaction
        addNewBranch("refs/heads/new-branch", "0a943a29376f2336b78312d99e65da17048951db");
    }

    private void addNewBranch(String refId, String hash) {
        testHelper.updateRef(meshManager.getLocalNode(), new UpdateRefParameters.Builder(repository,
                ProposedRefUpdate.toAdd(refId, hash))
                .build());
    }

    private void assertNodeState(MeshNode node, RpcReplicaState replicaStateConsistent) {
        assertThat("Node " + node, testHelper.getReplicaState(node, repository.getId()).getState(),
                is(replicaStateConsistent));
    }

    private void createInconsistentContentHashThroughRepair() {
        repairSource = getRemoteNode();
        repairTarget = meshManager.getLocalNode();
        remainingNode = meshManager.getNodes().stream()
                .filter(n -> !Objects.equals(repairSource, n))
                .filter(n -> !Objects.equals(repairTarget, n))
                .findFirst()
                .orElseThrow(() -> new IllegalStateException("No remaining node found"));

        UpdateRefParameters updateToBranch = new UpdateRefParameters.Builder(repository,
                ProposedRefUpdate.of("refs/heads/branch_mod_merge",
                        "6053a1eaa1c009dd11092d09a72f3c41af1b59ad", "9c05f43f859375e392d90d23a13717c16d0fdcda"))
                .build();

        UpdateRefParameters udpateToMaster = new UpdateRefParameters.Builder(repository,
                ProposedRefUpdate.of("refs/heads/master",
                        "0a943a29376f2336b78312d99e65da17048951db", "e0a3f6d8c804c375f49d7609328111e8dc75a4bc"))
                .build();

        //The goal is to introduce a state where repair forces to clear the stored content hashes, while there was
        //already a discrepancy between the stored content hashes and what's on disk.
        //
        // 1. Update a ref on all node without triggering hooks - this updates the repository on disk but does not
        //    update the stored content hash. This is the case that we still need to figure out why it happens
        //
        Stream.of(repairTarget, repairSource, remainingNode).forEach(node ->
                testHelper.updateRefNoHooks(node, udpateToMaster));
        //
        // 2. Update a different ref through a transaction while isolating the repair target. This applies this ref change
        //    through a transaction on the repair source and the remaining node. However, we also update the ref
        //    on the isolated repair target, causing its content hash to become out of date with the other nodes and
        //    triggering the repair. However, we also update the ref without triggering the hooks, which causes the repair
        //    not to detect any out of date refs and forcing the content hashes to be cleared on both the source and target.
        //
        withNodeOffline(repairTarget, () -> {
            testHelper.updateRef(repairSource, updateToBranch);
            testHelper.updateRefNoHooks(repairTarget, updateToBranch);
        });

        //
        // 3. Now we have the case where the refs on all nodes are the same, however, one node (the repair target) has
        //    an out of date content hash and will be repaired.
        //    Now we can trigger the repair.
        //
        ReplicaState state = testHelper.repair(repairTarget, repairSource, repairTarget, repository.getId());

        //
        // 4. Repair has detected no out of date refs, it cleared and recalculated the content hashes on the repair source
        //    and target. However, because the content hashes were out of sync with the repository on the remaining node,
        //    the repair source and target are now at the correct content hash and the remaining node is at a different
        //    content hash, even though now all are marked as consistent.
        assertEquals(CONSISTENT, state);
    }

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

    private void setForcedContentHashRecalculation(boolean enabled) {
        testHelper.setConfig("replication.content-hash-forced-recalculation-frequency", enabled ? "1" : "0");
    }

    private void updateRefWithoutHooksToRandomHash(Collection<MeshNode> nodes) {
        checkArgument(availableCommitHashes.size() >= nodes.size());

        nodes.forEach(node -> {
            UpdateRefParameters updateToBranch = new UpdateRefParameters.Builder(repository,
                    ProposedRefUpdate.of("refs/heads/branch_mod_merge",
                            "6053a1eaa1c009dd11092d09a72f3c41af1b59ad", availableCommitHashes.pop()))
                    .build();
            testHelper.updateRefNoHooks(node, updateToBranch);
        });
    }

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

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