package com.atlassian.stash.internal.pull;

import com.atlassian.bitbucket.concurrent.PullRequestLock;
import com.atlassian.bitbucket.concurrent.SimpleLockService;
import com.atlassian.hazelcast.test.HazelcastCluster;
import com.atlassian.stash.internal.repository.InternalRepository;
import com.atlassian.bitbucket.util.UncheckedOperation;
import com.hazelcast.core.HazelcastInstance;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.runners.MockitoJUnitRunner;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

import static org.junit.Assert.assertEquals;
import static org.mockito.Mockito.*;


@RunWith(MockitoJUnitRunner.class)
public class HazelcastPullRequestStateGuardTest {

    @ClassRule
    public static HazelcastCluster cluster = new HazelcastCluster.Builder().size(2).build();

    private HazelcastPullRequestStateGuard commentStateGuard;
    @Mock
    private InternalPullRequest pullRequest;
    @Mock
    private UncheckedOperation<Void> operation;

    @Before
    public void setup() {
        HazelcastInstance hazelcastInstance = cluster.getNode(1);
        PullRequestLock lock = new SimpleLockService().getPullRequestLock("state-guard-test");
        commentStateGuard = new HazelcastPullRequestStateGuard(lock, hazelcastInstance.<Long, Integer>getMap("GUARD"));

        when(pullRequest.getId()).thenReturn(1L);
        when(pullRequest.getGlobalId()).thenReturn(1L);

        InternalRepository mockRepository = mock(InternalRepository.class);
        when(mockRepository.getId()).thenReturn(1);

        InternalPullRequestRef mockRef = mock(InternalPullRequestRef.class);
        when(mockRef.getRepository()).thenReturn(mockRepository);

        when(pullRequest.getToRef()).thenReturn(mockRef);
    }

    @Test
    public void testPullRequestAssumedOutdated() {
        commentStateGuard.updated(pullRequest);
        commentStateGuard.processIfUpdated(pullRequest, operation);

        verify(operation).perform();
    }

    @Test
    public void testUptodatePullRequest() {
        commentStateGuard.updated(pullRequest);
        commentStateGuard.processIfUpdated(pullRequest, operation);
        commentStateGuard.processIfUpdated(pullRequest, operation);

        // should only be called once
        verify(operation).perform();

        commentStateGuard.updated(pullRequest);
        commentStateGuard.processIfUpdated(pullRequest, operation);

        verify(operation, times(2)).perform();
    }

    @Test(timeout = 10000)
    public void testMulithreadedWithoutRescope() throws Exception {
        doTestMultithreaded(false);
    }

    @Test(timeout = 10000)
    public void testMulithreadedWithRescope() throws Exception {
        doTestMultithreaded(true);
    }

    private void doTestMultithreaded(final boolean rescopeWhileBlocked) throws Exception {
        final CountDownLatch threadsReady = new CountDownLatch(2);
        final CountDownLatch updateStarted = new CountDownLatch(1);
        final CountDownLatch updateFinished = new CountDownLatch(1);
        final AtomicInteger operationsPerformed = new AtomicInteger();

        commentStateGuard.updated(pullRequest);

        // start a thread that enters the synchronized update block inside guard, then blocks
        new Thread(() -> {
            threadsReady.countDown();
            commentStateGuard.processIfUpdated(pullRequest, () -> {
                if (rescopeWhileBlocked) {
                    // simulate a rescope - mark as updated
                    commentStateGuard.updated(pullRequest);
                }
                updateStarted.countDown();
                try {
                    updateFinished.await();
                    operationsPerformed.incrementAndGet();
                } catch (InterruptedException e) {
                    // ignore
                }
                return null;
            });
        }).start();

        // start a second thread that calls guard.processIfUpdated while thread1 is in updating
        Thread thread2 = new Thread(() -> {
            threadsReady.countDown();
            try {
                updateStarted.await();
            } catch (InterruptedException e) {
                // ignore
            }
            // should block until thread1 finishes
            commentStateGuard.processIfUpdated(pullRequest, () -> {
                operationsPerformed.incrementAndGet();
                return null;
            });
        });
        thread2.start();

        // wait until both threads have started and are ready to call processIfUpdated
        threadsReady.await();
        updateStarted.await();

        // let the first thread process
        updateFinished.countDown();

        // wait for the second thread to complete
        thread2.join();

        if (!rescopeWhileBlocked) {
            // the second thread shouldn't have triggered an update
            assertEquals(1, operationsPerformed.get());
        } else {
            assertEquals(2, operationsPerformed.get());
        }
    }
}
