package com.atlassian.bitbucket.mesh;

import com.atlassian.bitbucket.mesh.git.ref.Branch;
import com.atlassian.bitbucket.mesh.git.ref.ProposedRefUpdate;
import com.atlassian.bitbucket.mesh.git.ref.Ref;
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.*;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcRepositoryConsistencySummary;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcRepositoryIntegrityIssue;
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 io.grpc.Status;
import io.grpc.StatusRuntimeException;
import org.junit.Before;
import org.junit.Test;

import javax.annotation.Resource;
import java.time.Duration;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

import static com.atlassian.bitbucket.mesh.rpc.v1.RpcReplicaState.REPLICA_STATE_CONSISTENT;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.*;
import static org.hamcrest.collection.IsEmptyCollection.empty;
import static org.junit.Assert.*;

public class VerifyConsistencyAndIntegrityChecksTest extends AbstractWiredTest {

    @Resource
    private MeshClusterManager clusterManager;
    @Resource
    private MeshManager meshManager;
    private List<MeshNode> remoteNodes;
    @Repo
    private ManagedRepository repository;
    @Resource
    private RepositoryAttributeManager repositoryAttributeManager;
    @Resource
    private RepositoryHashService repositoryHashService;
    @Resource
    private RepositoryManager repositoryManager;
    @Resource
    private TestHelper testHelper;

    @Before
    public void setUp() {
        remoteNodes = meshManager.getNodes().stream()
                .filter(MeshNode::isRemote)
                .collect(Collectors.toList());
        assertThat("At least 2 remote nodes are required for this test", remoteNodes.size(), greaterThanOrEqualTo(2));
    }

    @Test
    public void testBulkVerifyIntegrityCheckAllConsistent() {
        List<RpcRepositoryIntegrityIssue> issues = testHelper.verifyIntegrity(
                remoteNodes.get(0), repository.getId(), 0L);

        //All should be consistent
        assertThat(issues, empty());
    }

    @Test
    public void testVerifyConsistencyAllConsistent() {
        //Have consistentNode coordinate the consistency check
        RpcRepositoryConsistencySummary summary = testHelper.verifyConsistency(remoteNodes.get(0), repository.getId());

        assertNotNull(summary);
        assertThat(summary.getContentHash(), equalTo(repositoryHashService.getContentHash(repository)));
        assertThat(summary.getVersion(), greaterThan(1L)); //At least 2 after creation
        assertThat(summary.getUnavailableReplicasList(), empty());
        assertThat(summary.getInconsistentReplicasList(), empty());
    }

    @DirtiesRepo
    @Test
    public void testVerifyConsistencyLeaderConsistent() throws Exception {
        //Create the branch with on remoteNodes[1] with remoteNodes[0] offline.
        //Only remoteNodes[0] should be inconsistent.
        String branchId = "refs/heads/this-is-a-new-branch";
        MeshNode inconsistentNode = remoteNodes.get(0);
        MeshNode consistentNode = remoteNodes.get(1);
        Branch branch = clusterManager.getNodeById(inconsistentNode.getId())
                .whileOffline(() -> testHelper.createBranchWithNewCommit(consistentNode, repository.getId(), branchId));

        //Have consistentNode coordinate the consistency check
        RpcRepositoryConsistencySummary summary = testHelper.verifyConsistency(consistentNode, repository.getId());

        assertNotNull(summary);
        assertThat(summary.getContentHash(), equalTo(repositoryHashService.getContentHash(repository)));
        assertThat(summary.getVersion(), greaterThan(2L)); //At least 2 after creation + at least one write
        assertThat(summary.getUnavailableReplicasList(), empty());
        assertThat(summary.getInconsistentReplicasList(), contains(inconsistentNode.getId()));

        awaitTrue("The inconsistent node did not repair itself", Duration.ofSeconds(5L), () -> {
            try {
                Ref ref = testHelper.resolveRef(inconsistentNode, repository.getId(), branchId);
                return ref != null && ref.getLatestCommit().equals(branch.getLatestCommit());
            } catch (RuntimeException e) {
                return false;
            }
        });
    }

    @DirtiesRepo
    @Test
    public void testVerifyConsistencyLeaderInconsistent() throws Exception {
        //Create the branch with on remoteNodes[1] with remoteNodes[0] offline.
        //Only remoteNodes[0] should be inconsistent.
        String branchId = "refs/heads/this-is-a-new-branch";
        MeshNode inconsistentNode = remoteNodes.get(0);
        MeshNode consistentNode = remoteNodes.get(1);
        Branch branch = clusterManager.getNodeById(inconsistentNode.getId())
                .whileOffline(() -> testHelper.createBranchWithNewCommit(consistentNode, repository.getId(), branchId));

        //Have inconsistentNode coordinate the consistency check
        RpcRepositoryConsistencySummary summary = testHelper.verifyConsistency(inconsistentNode, repository.getId());

        assertNotNull(summary);
        assertThat(summary.getContentHash(), equalTo(repositoryHashService.getContentHash(repository)));
        assertThat(summary.getVersion(), greaterThan(2L)); //At least 2 after creation + at least one write
        assertThat(summary.getUnavailableReplicasList(), empty());
        assertThat(summary.getInconsistentReplicasList(), contains(inconsistentNode.getId()));

        waitForAllNodesToBecomeConsistent(repository.getId());
    }

    @Test
    public void testVerifyConsistencyAllMissingRepository() {
        String missingId = ManagedRepository.formatId(1, "99999999991111111111", "792342");

        StatusRuntimeException exception = assertThrows(StatusRuntimeException.class,
                () -> testHelper.verifyConsistency(meshManager.getLocalNode(), missingId));

        assertEquals(Status.Code.NOT_FOUND, exception.getStatus().getCode());
    }

    @Test
    public void testVerifyConsistencyFollowerMissingRepository() throws Exception {
        //Create the branch with on remoteNodes[1] with remoteNodes[0] offline.
        //Only remoteNodes[0] should be inconsistent.
        MeshNode inconsistentNode = remoteNodes.get(0);
        MeshNode consistentNode = remoteNodes.get(1);
        String newRepositoryId = ManagedRepository.formatId(1, "99999999991111111111", "1299");
        try {
            clusterManager.getNodeById(inconsistentNode.getId())
                    .whileOffline(() -> {
                        testHelper.createRepository(consistentNode, new CreateRepositoryParameters.Builder(newRepositoryId)
                                .build());
                        return null;
                    });

            //Have consistentNode coordinate the consistency check
            RpcRepositoryConsistencySummary summary = testHelper.verifyConsistency(consistentNode, newRepositoryId);

            assertNotNull(summary);
            assertThat(summary.getContentHash(), equalTo(RefHasher.NULL_HASH)); //empty
            assertThat(summary.getVersion(), equalTo(2L));
            assertThat(summary.getUnavailableReplicasList(), empty());
            assertThat(summary.getInconsistentReplicasList(), contains(inconsistentNode.getId()));

            //Verify that the consistent node will attempt to repair itself
            waitForAllNodesToBecomeConsistent(newRepositoryId);
        } finally {
            testHelper.deleteRepository(consistentNode, newRepositoryId);
        }
    }

    @Test
    public void testVerifyConsistencyLeaderMissingRepository() throws Exception {
        //Create the branch with on remoteNodes[1] with remoteNodes[0] offline.
        //Only remoteNodes[0] should be inconsistent.
        MeshNode inconsistentNode = remoteNodes.get(0);
        MeshNode consistentNode = remoteNodes.get(1);
        String newRepositoryId = ManagedRepository.formatId(12, "99999999992222222222", "2299");
        try {
            clusterManager.getNodeById(inconsistentNode.getId())
                    .whileOffline(() -> {
                        testHelper.createRepository(consistentNode, new CreateRepositoryParameters.Builder(newRepositoryId)
                                .build());
                        return null;
                    });

            //Have inconsistentNode coordinate the consistency check
            RpcRepositoryConsistencySummary summary = testHelper.verifyConsistency(inconsistentNode, newRepositoryId);

            assertNotNull(summary);
            assertThat(summary.getContentHash(), equalTo(RefHasher.NULL_HASH)); //empty
            assertThat(summary.getVersion(), equalTo(2L));
            assertThat(summary.getUnavailableReplicasList(), empty());
            assertThat(summary.getInconsistentReplicasList(), contains(inconsistentNode.getId()));

            //Verify that the consistent node will attempt to repair itself
            awaitTrue("The inconsistent node did not repair itself", Duration.ofSeconds(5L), () -> {
                try {
                    testHelper.getDetails(inconsistentNode, newRepositoryId);
                    return true;
                } catch (RuntimeException e) {
                    return false;
                }
            });
        } finally {
            testHelper.deleteRepository(consistentNode, newRepositoryId);
        }
    }

    @Test
    public void testVerifyConsistencyMajorityMissingAtSameVersion() throws Exception {
        String newRepositoryId = ManagedRepository.formatId(1, "99999999991111111112", "1300");
        try {
            //Only create repository locally, other nodes will not have the repository
            localNode.whileOffline(() -> {
                testHelper.createRepository(meshManager.getLocalNode(), new CreateRepositoryParameters.Builder(newRepositoryId).build());
                String commit = testHelper.createCommit(meshManager.getLocalNode(), newRepositoryId);
                testHelper.updateRef(meshManager.getLocalNode(),
                        new UpdateRefParameters.Builder(repositoryManager.getById(newRepositoryId), ProposedRefUpdate.toAdd("refs/heads/main", commit)).build());
                //Make sure the version is the same as the "missing" version, to verify the behavior of selecting the
                //correct consensus version
                repositoryAttributeManager.setVersion(newRepositoryId, 1L);
                repositoryHashService.getContentHash(repositoryManager.getById(newRepositoryId));
                return null;
            });

            //Have the local node run the consistency check
            RpcRepositoryConsistencySummary summary =
                    testHelper.verifyConsistency(meshManager.getLocalNode(), newRepositoryId);

            assertNotNull(summary);
            String contentHash = repositoryHashService.getContentHash(repositoryManager.getById(newRepositoryId));
            assertThat(contentHash, is(not(RefHasher.NULL_HASH)));
            assertThat(summary.getContentHash(), equalTo(contentHash));
            assertThat(summary.getVersion(), equalTo(1L));
            assertThat(summary.getInconsistentReplicasCount(), equalTo(2));
            assertThat(summary.getInconsistentReplicasList(), containsInAnyOrder(remoteNodes.stream().map(MeshNode::getId).toArray(String[]::new)));
            assertThat(summary.getUnavailableReplicasList(), empty());

            //Verify that the consistent node will attempt to repair itself
            waitForAllNodesToBecomeConsistent(newRepositoryId);
            remoteNodes.forEach(remoteNode ->
                    assertEquals(contentHash, testHelper.getReplicaState(remoteNode, newRepositoryId).getContentHash()));
        } finally {
            testHelper.deleteRepository(meshManager.getLocalNode(), newRepositoryId);
        }
    }

    private void waitForAllNodesToBecomeConsistent(String repositoryId) {
        Set<MeshNode> inconsistentNodes = new HashSet<>(meshManager.getNodes());
        awaitTrue("At least one replica was inconsistent!", Duration.ofSeconds(5L), () -> {
            Iterator<MeshNode> it = inconsistentNodes.iterator();
            while (it.hasNext()) {
                try {
                    if (testHelper.getReplicaState(it.next(), repositoryId).getState() == REPLICA_STATE_CONSISTENT) {
                        it.remove();
                    }
                } catch (RuntimeException e) {
                    //ignore
                }
            }
            return inconsistentNodes.isEmpty();
        });
    }
}
