package com.atlassian.bitbucket.mesh.transaction;

import com.atlassian.bitbucket.mesh.AbstractGrpcTest;
import com.atlassian.bitbucket.mesh.git.GitAgent;
import com.atlassian.bitbucket.mesh.git.GrpcTransactionService;
import com.atlassian.bitbucket.mesh.git.builder.GitProcessBuilderFactory;
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.network.UpdateTopologyParameters;
import com.atlassian.bitbucket.mesh.repository.ManagedRepository;
import com.atlassian.bitbucket.mesh.repository.RepositoryManager;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcPerson;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcPrepareTransactionResponse;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcTransaction;
import com.atlassian.bitbucket.mesh.rpc.v1.TransactionServiceGrpc;
import com.atlassian.bitbucket.mesh.test.annotation.DirtiesRepo;
import com.atlassian.bitbucket.mesh.test.annotation.Repo;
import com.atlassian.bitbucket.mesh.util.Person;
import io.grpc.ManagedChannel;
import org.junit.*;

import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

public abstract class AbstractPreparedActionWriteTest extends AbstractGrpcTest {

    static final long TX_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(3);
    static final String LEADER_ID = "1eade4";

    static final RpcPerson USER = new Person("John Doe", "johnd@email.com").toPerson();

    private static final AtomicLong nextTransactionId = new AtomicLong(1L);

    @Rule
    public final LogListener txServiceLogListener = new LogListener(GrpcTransactionService.class);
    @Rule
    public final LogListener remoteTxLeaderLogListener =
            new LogListener("com.atlassian.bitbucket.mesh.git.RemoteTransactionLeader");

    @Resource
    protected GitAgent agent;
    @Resource
    protected GitProcessBuilderFactory builderFactory;
    @Resource
    protected ManagedChannel managedChannel;
    @Resource
    protected MeshManager meshManager;
    @Resource
    protected RepositoryManager repositoryManager;
    @Repo
    protected ManagedRepository repository;

    @Before
    public void setupLeader() {
        MeshNode localNode = meshManager.getLocalNode();
        meshManager.updateTopology(new UpdateTopologyParameters.Builder()
                .nodes(localNode, new MeshNode.Builder(LEADER_ID)
                        .name("leader")
                        .rpcUrl("http://localhost:56987")
                        .build())
                .build());
    }

    @After
    public void tearDownLeader() {
        MeshNode localNode = meshManager.getLocalNode();
        meshManager.updateTopology(new UpdateTopologyParameters.Builder()
                .nodes(localNode)
                .build());
    }

    @Ignore("This test (in both subclasses) is currently too flaky to be part of our main builds")
    @Test
    public void testTransactAbortsOnFollowerError() {
        //Try to replicate a transaction that triggers an error on the transaction follower. This error should be
        //handled and translated into a NO vote.
        BlockingReplicaTransaction tx = startReplicaTransaction();
        RpcTransaction transaction = createTransactionThatTriggersError(RpcTransaction.newBuilder()
                .setId(nextTransactionId())
                .setPerson(USER)
                .setRepository(repository.getId())
                .setTimestamp(System.currentTimeMillis()))
                .build();

        RpcPrepareTransactionResponse prepareResponse = tx.prepare(transaction);

        assertEquals(RpcPrepareTransactionResponse.Vote.NO, prepareResponse.getVote());
        remoteTxLeaderLogListener.assertWarn("Failed to prepare transaction");
    }

    @Test
    public void testTransactAbortsOnLeaderError() {
        BlockingReplicaTransaction tx = startReplicaTransaction();
        RpcTransaction transaction = createTransaction(RpcTransaction.newBuilder()
                .setId(nextTransactionId())
                .setPerson(USER)
                .setRepository(repository.getId())
                .setTimestamp(System.currentTimeMillis()))
                .build();

        RpcPrepareTransactionResponse prepareResponse = tx.prepare(transaction);
        assertEquals(RpcPrepareTransactionResponse.Vote.YES, prepareResponse.getVote());

        tx.cancel("Simulated leader error", new IllegalStateException("Simulated error on leader"));

        //validate that the transaction has not been applied
        assertFalse(isApplied(transaction));
    }

    @Test
    public void testTransactPrepareAndAbort() {
        BlockingReplicaTransaction tx = startReplicaTransaction();
        RpcTransaction transaction = createTransaction(RpcTransaction.newBuilder()
                .setId(nextTransactionId())
                .setPerson(USER)
                .setRepository(repository.getId())
                .setTimestamp(System.currentTimeMillis()))
                .build();

        RpcPrepareTransactionResponse prepareResponse = tx.prepare(transaction);
        assertEquals(RpcPrepareTransactionResponse.Vote.YES, prepareResponse.getVote());

        tx.rollBack();
        tx.summarize(false, "abcdef");

        //validate that the transaction has not been applied
        assertFalse(isApplied(transaction));
    }

    @DirtiesRepo
    @Test
    public void testTransactPrepareAndCommit() {
        BlockingReplicaTransaction tx = startReplicaTransaction();
        RpcTransaction transaction = createTransaction(RpcTransaction.newBuilder()
                .setId(nextTransactionId())
                .setPerson(USER)
                .setRepository(repository.getId())
                .setTimestamp(System.currentTimeMillis()))
                .build();

        RpcPrepareTransactionResponse prepareResponse = tx.prepare(transaction);
        assertEquals(RpcPrepareTransactionResponse.Vote.YES, prepareResponse.getVote());

        tx.commit();
        tx.summarize(false, "abcdef");

        //validate that the transaction has been applied
        assertTrue(isApplied(transaction));
    }

    static String nextTransactionId() {
        return Transaction.formatId(LEADER_ID, Long.toHexString(nextTransactionId.getAndIncrement()));
    }

    protected abstract RpcTransaction.Builder createTransaction(RpcTransaction.Builder builder);

    protected abstract RpcTransaction.Builder createTransactionThatTriggersError(RpcTransaction.Builder builder);

    protected abstract boolean isApplied(RpcTransaction transaction);

    BlockingReplicaTransaction startReplicaTransaction() {
        BlockingReplicaTransaction replicaTransaction = new BlockingReplicaTransaction(repository);
        TransactionServiceGrpc.newStub(managedChannel)
                .withDeadlineAfter(TX_TIMEOUT_MS + 1000, TimeUnit.MILLISECONDS)
                .transact(replicaTransaction);
        assertTrue("ReplicaTransaction was not started", replicaTransaction.isStarted());
        return replicaTransaction;
    }
}
