package com.atlassian.bitbucket.mesh;

import com.atlassian.bitbucket.mesh.git.exception.ObjectNotFoundException;
import com.atlassian.bitbucket.mesh.git.ref.Branch;
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.RpcGitError;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcGitErrorUtils;
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.StatusRuntimeException;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Resource;
import java.util.List;
import java.util.stream.Collectors;

import static com.atlassian.bitbucket.mesh.git.GitObjectType.COMMIT;
import static org.hamcrest.CoreMatchers.containsString;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.collection.IsEmptyCollection.empty;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertSame;

public class ReplicationTest extends AbstractWiredTest {

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

    @Resource
    private MeshClusterManager clusterManager;
    @Resource
    private MeshManager meshManager;
    @Repo
    private ManagedRepository repository;
    @Resource
    private RepositoryManager repositoryManager;
    @Resource
    private TestHelper testHelper;

    @DirtiesRepo
    @Test
    public void testFailedReplication() throws Exception {
        MeshNode remoteReplica = meshManager.getNodes().stream()
                .filter(MeshNode::isRemote)
                .findAny()
                .orElse(null);
        //Verify that there is indeed a cluster with remote nodes
        assertNotNull("There are no remote Mesh nodes", remoteReplica);

        //Create the branch on an offline node
        String branch = "refs/heads/already-created";
        clusterManager.getNodeById(remoteReplica.getId())
                .whileOffline(() -> testHelper.createBranchWithNewCommit(remoteReplica, repository.getId(), branch));

        try {
            //What happens from here is dependent on the QuorumMode
            testHelper.createBranchWithNewCommit(meshManager.getLocalNode(), repository.getId(), branch);

            //ALWAYS: Even though remoteReplica vetoes, the transaction should still be committed
            assertSame(COMMIT, testHelper.getObjectType(meshManager.getLocalNode(), repository, branch));
        } catch (StatusRuntimeException e) {
            //UNANIMOUS: remoteReplica votes no, so the branch creation should be rolled back and a
            //StatusRuntimeException should be thrown
            RpcGitError error = RpcGitErrorUtils.fromThrowable(e);
            assertNotNull("Missing error details", error);
            assertThat(error.getFailedCommand().getStderr(), containsString("ref updates aborted by hook"));
        }
    }

    @Test
    public void testForkRepository() {
        String forkId = repository.getId() + "972934";

        Repository fork = repositoryManager.create(new CreateRepositoryParameters.Builder(forkId)
                .origin(repository)
                .build());

        try {
            testHelper.assertRefsInSync(meshManager.getNodes(), fork.getId());
        } finally {
            repositoryManager.delete(new DeleteRepositoryParameters.Builder(fork).build());
        }
    }

    @DirtiesRepo
    @Test
    public void testUpdateRefFetchesMissingObjects() {
        MeshNode remoteReplica = meshManager.getNodes().stream()
                .filter(MeshNode::isRemote)
                .findFirst()
                .orElse(null);
        //Verify that there is indeed a cluster with remote nodes
        assertNotNull("There are no remote Mesh nodes", remoteReplica);

        //Introduce new objects to a replica repository
        Branch branch = testHelper.createBranchWithNewCommit(
                remoteReplica, repository.getId(), "refs/heads/'justin'-time-fetch");

        //The reference-transaction hook should trigger an update-ref on all replicas. These should all fail
        //initially because none of the replicas have the objects yet. They should then fetch the missing
        //objects and retry the update, which should then succeed
        List<MeshNode> unsynced = meshManager.getNodes().stream()
                .filter(replica -> {
                    try {
                        return testHelper.getObjectType(replica, repository, branch.getId()) != COMMIT;
                    } catch (ObjectNotFoundException e) {
                        log.warn("[{}] {}", replica.getName(), e.getMessage());
                        return false;
                    }
                })
                .collect(Collectors.toList());
        assertThat(String.format("Replicas %s were not synced!", unsynced), unsynced, empty());
    }
}
