package com.atlassian.bitbucket.mesh.transaction;

import com.atlassian.bitbucket.mesh.config.ReplicationConfig;
import com.atlassian.bitbucket.mesh.execution.GrpcExecutionManager;
import com.atlassian.bitbucket.mesh.git.ref.RefUpdate;
import com.atlassian.bitbucket.mesh.junit.LogListener;
import com.atlassian.bitbucket.mesh.repository.*;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcTransaction;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcUpdateRefsRequest;
import com.atlassian.bitbucket.mesh.test.ExecutionContextRule;
import com.atlassian.bitbucket.mesh.util.Person;
import com.google.common.collect.ImmutableSet;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.InOrder;
import org.mockito.Mock;
import org.mockito.Spy;
import org.mockito.junit.MockitoJUnitRunner;

import static java.util.Optional.empty;
import static org.junit.Assert.*;
import static org.mockito.Mockito.*;

@RunWith(MockitoJUnitRunner.class)
public class CoordinatingReferenceTransactionListenerTest {

    private static final RefUpdate REF_UPDATE = RefUpdate.of("refs/heads/branch",
            "dfa28787945c3848bd791490eb487734440f56d7", "2c617209c6980caf9b0c38004cd9010e4980ae9e");
    private static final String REPOSITORY_ID = ManagedRepository.formatId(1, "hierarchy-id", "local-id");
    private static final Person USER = new Person("user", null);

    @Rule
    public final ExecutionContextRule executionContextRule = new ExecutionContextRule(new GrpcExecutionManager());
    @Rule
    public final LogListener logListener = new LogListener(CoordinatingReferenceTransactionListener.class);
    @Rule
    @Spy
    public final StubTransactionManager transactionManager = new StubTransactionManager();

    @Mock
    private ReplicationConfig config;
    //This is done using extraInterfaces to allow reassigning this field with repository mocks of other types
    @Mock(extraInterfaces = ManagedRepository.class)
    private Repository repository;
    @Mock
    private RepositoryHashService repositoryHashService;
    private CoordinatingReferenceTransactionListener listener;

    @Before
    public void setUp() {
        when(repository.accept(any())).thenAnswer(invocation -> {
            RepositoryVisitor<?> visitor = invocation.getArgument(0);

            return visitor.visit((ManagedRepository) repository);
        });
        when(repository.getId()).thenReturn(REPOSITORY_ID);
        when(repositoryHashService.calculateTransactionHash(anySet())).thenAnswer(invocation ->
                Long.toHexString(invocation.getArgument(0).hashCode()));

        listener = new CoordinatingReferenceTransactionListener(
                config, executionContextRule.getExecutionManager(), repositoryHashService, transactionManager);
    }

    @Test
    public void testAbort() {
        transactionManager.stubCoordinatorVote(false);

        ReferenceTransaction refTx = invokeHook(ReferenceTransactionState.PREPARED);
        assertTransaction(TransactionState.ROLLING_BACK);
        verify(refTx).reject(anyString());

        refTx = invokeHook(ReferenceTransactionState.ABORTED);
        assertTransaction(TransactionState.ROLLED_BACK);
        verify(refTx, never()).reject(anyString());
    }

    @Test
    public void testAbortWithoutAbortedCallback() {
        //Call the PREPARED callback, then close the execution context
        executionContextRule.runInNewContext(() -> {
            transactionManager.stubCoordinatorVote(false);

            ReferenceTransaction refTx = invokeHook(ReferenceTransactionState.PREPARED);
            assertTransaction(TransactionState.ROLLING_BACK);
            verify(refTx).reject(anyString());

            //Simulate the transaction closing at execution context cleanup
            transactionManager.getCurrentTransaction().ifPresent(Transaction::close);
        });

        //Verify that the transaction has been fully aborted even if we never received the ABORTED callback
        assertTransaction(TransactionState.ROLLED_BACK);
        logListener.assertInfo(
                "Transaction rolled back as expected but no ABORTED reference-transaction hook has been received yet");
    }

    @Test
    public void testAbortWithoutPrepareIsLogged() {
        ReferenceTransaction refTx = invokeHook(ReferenceTransactionState.ABORTED);
        verify(refTx, never()).reject(anyString());

        assertEquals(empty(), transactionManager.getCurrentTransaction());
        logListener.assertWarn("ABORTED callback received but no transaction has been PREPARED");
    }

    @Test
    public void testCommit() {
        transactionManager.stubCoordinatorVote(true);

        when(repositoryHashService.getContentHash(same(repository))).thenReturn("original");

        ReferenceTransaction refTx = invokeHook(ReferenceTransactionState.PREPARED);
        assertTransaction(TransactionState.COMMITTING);
        verify(refTx, never()).reject(anyString());

        //Verify the ref update is prepared with the TransactionManager before the content hash is captured.
        //This avoids a race condition where multiple ref updates can capture the same content hash and then
        //one (or more) will fail to update because the expected content hash is stale
        InOrder ordered = inOrder(repositoryHashService, transactionManager);
        ordered.verify(transactionManager).prepared(any());
        ordered.verify(repositoryHashService).getContentHash(same(repository));

        refTx = invokeHook(ReferenceTransactionState.COMMITTED);
        assertTransaction(TransactionState.COMMITTED);
        verify(refTx, never()).reject(anyString());

        verify(repositoryHashService).compareAndUpdateContentHash(same(repository), eq("original"), anyString());
    }

    @Test
    public void testCommitForcedContentHashRecalculation() {
        transactionManager.stubCoordinatorVote(true);

        when(repositoryHashService.getContentHash(same(repository))).thenReturn("original");

        ReferenceTransaction refTx = invokeHook(ReferenceTransactionState.PREPARED);
        assertTransaction(TransactionState.COMMITTING);
        verify(refTx, never()).reject(anyString());

        //Verify the ref update is prepared with the TransactionManager before the content hash is captured.
        //This avoids a race condition where multiple ref updates can capture the same content hash and then
        //one (or more) will fail to update because the expected content hash is stale
        InOrder ordered = inOrder(repositoryHashService, transactionManager);
        ordered.verify(transactionManager).prepared(any());
        ordered.verify(repositoryHashService).getContentHash(same(repository));

        refTx = invokeHook(ReferenceTransactionState.COMMITTED);
        assertTransaction(TransactionState.COMMITTED);
        verify(refTx, never()).reject(anyString());

        verify(repositoryHashService).compareAndUpdateContentHash(same(repository), eq("original"), anyString());
    }

    @Test
    public void testCommitWithoutCommittedCallback() {
        assertThrows(TransactionCommitFailedException.class, () -> executionContextRule.runInNewContext(() -> {
            //Call the PREPARED callback, then close the execution context
            transactionManager.stubCoordinatorVote(true);

            invokeHook(ReferenceTransactionState.PREPARED);
            assertTransaction(TransactionState.COMMITTING);

            //Simulate the transaction closing at execution context cleanup
            transactionManager.getCurrentTransaction().ifPresent(Transaction::close);
        }));
    }

    @Test
    public void testCommitWithoutPrepareIsLogged() {
        ReferenceTransaction refTx = invokeHook(ReferenceTransactionState.COMMITTED);
        verify(refTx, never()).reject(anyString());

        assertEquals(empty(), transactionManager.getCurrentTransaction());
    }

    @Test
    public void testOnReferenceTransactionIgnoresBitbucketServerRepository() {
        repository = mock(BitbucketServerRepository.class);
        when(repository.accept(any())).thenAnswer(invocation -> {
            RepositoryVisitor<?> visitor = invocation.getArgument(0);

            return visitor.visit((BitbucketServerRepository) repository);
        });

        invokeHook(ReferenceTransactionState.PREPARED);

        verify(repository).accept(notNull());
        verifyNoInteractions(repositoryHashService, transactionManager);
    }

    @Test
    public void testOnReferenceTransactionIgnoresPathRepository() {
        repository = mock(PathRepository.class);
        when(repository.accept(any())).thenAnswer(invocation -> {
            RepositoryVisitor<?> visitor = invocation.getArgument(0);

            return visitor.visit((PathRepository) repository);
        });

        invokeHook(ReferenceTransactionState.PREPARED);

        verify(repository).accept(notNull());
        verifyNoInteractions(repositoryHashService, transactionManager);
    }

    @Test
    public void testOnReferenceTransactionIgnoresSharedObjectsRepository() {
        repository = mock(SharedObjectsRepository.class);
        when(repository.accept(any())).thenAnswer(invocation -> {
            RepositoryVisitor<?> visitor = invocation.getArgument(0);

            return visitor.visit((SharedObjectsRepository) repository);
        });

        invokeHook(ReferenceTransactionState.PREPARED);

        verify(repository).accept(notNull());
        verifyNoInteractions(repositoryHashService, transactionManager);
    }

    @Test
    public void testPreparedRpcTransaction() {
        invokeHook(ReferenceTransactionState.PREPARED);

        RpcTransaction rpcTransaction = transactionManager.getPreparedRpcTransaction();
        assertNotNull(rpcTransaction);
        assertEquals(USER.toPerson(), rpcTransaction.getPerson());
        assertTrue(rpcTransaction.hasUpdateRefs());

        RpcUpdateRefsRequest updateRefsRequest = rpcTransaction.getUpdateRefs();
        assertEquals(REPOSITORY_ID, updateRefsRequest.getRepository());
        assertEquals(1, updateRefsRequest.getRefUpdatesCount());
        assertEquals(REF_UPDATE.toProposedRefUpdate(), updateRefsRequest.getRefUpdates(0));
    }

    @Test
    public void testUnexpectedAbortedCallbackThrows() {
        transactionManager.stubCoordinatorVote(false);

        invokeHook(ReferenceTransactionState.PREPARED);
        assertTransaction(TransactionState.ROLLING_BACK);

        assertThrows(TransactionRollbackFailedException.class, () -> invokeHook(ReferenceTransactionState.COMMITTED));
    }

    @Test
    public void testUnexpectedCommittedCallbackThrows() {
        transactionManager.stubCoordinatorVote(true);

        invokeHook(ReferenceTransactionState.PREPARED);
        assertTransaction(TransactionState.COMMITTING);

        assertThrows(TransactionCommitFailedException.class, () -> invokeHook(ReferenceTransactionState.ABORTED));
    }

    private void assertTransaction(TransactionState state) {
        Transaction transaction = state.isFinished() ?
                transactionManager.getLastTransaction() :
                transactionManager.getCurrentTransaction().orElse(null);

        assertNotNull("Transaction not found", transaction);
        assertEquals(state, transaction.getState());
    }

    private ReferenceTransaction invokeHook(ReferenceTransactionState state) {
        ReferenceTransaction referenceTransaction = mock(ReferenceTransaction.class);
        when(referenceTransaction.getRepository()).thenReturn(repository);
        when(referenceTransaction.getRefUpdates()).thenReturn(ImmutableSet.of(REF_UPDATE));
        when(referenceTransaction.getState()).thenReturn(state);
        when(referenceTransaction.getUser()).thenReturn(USER);

        listener.onReferenceTransaction(referenceTransaction);
        return referenceTransaction;
    }
}