package com.atlassian.bitbucket.mesh.coordination;

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.repository.Repository;
import com.google.common.collect.ImmutableList;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
import org.springframework.scheduling.TaskScheduler;

import javax.annotation.Nonnull;

import java.time.Duration;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.atomic.AtomicInteger;

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

@RunWith(MockitoJUnitRunner.class)
public class DefaultWriteTokenRegistryTest {

    private static final long TIMESTAMP = 162391L;

    @Rule
    public final LogListener logListener = new LogListener(DefaultWriteTokenRegistry.class);
    @Mock
    private MeshNode localNode;
    @Mock
    private LogicalClock logicalClock;
    @Mock
    private MeshManager meshManager;
    private DefaultWriteTokenRegistry registry;
    @Mock
    private MeshNode remoteNode;
    @Mock
    private Repository repository;
    @Mock
    private TaskScheduler taskScheduler;

    @Before
    public void setUp() {
        when(repository.getSlug()).thenReturn("repository-slug");
        ScheduledFuture future = mock(ScheduledFuture.class);
        when(taskScheduler.scheduleAtFixedRate(any(Runnable.class), any())).thenReturn(future);
        when(logicalClock.tick()).thenReturn(TIMESTAMP);
        when(meshManager.getLocalNode()).thenReturn(localNode);
        when(localNode.getId()).thenReturn("local");
        when(remoteNode.getId()).thenReturn("remote");

        registry = new DefaultWriteTokenRegistry(logicalClock, meshManager, taskScheduler, Duration.ofMillis(25L));
    }

    @Test
    public void testConcurrentRequestsInPriorityOrder() {
        GrantHandler handler1 = new GrantHandler();
        GrantHandler handler2 = new GrantHandler();
        try (WriteToken.Promise promise1 = registry.acquireToken(repository, "tx1", 1000L, handler1)) {
            //Token should be held by tx1
            assertEquals("tx1", registry.getHolder(repository));

            //Use the same timestamp to trigger sorting on secondary key - transactionId
            try (WriteToken.Promise ignored = registry.acquireToken(repository, "tx2", 1000L, handler2)) {
                //Token should still be held by tx1
                assertEquals("tx1", registry.getHolder(repository));

                assertFalse(handler2.isGranted());

                //Now simulate that tx1 releases the token
                promise1.close();

                assertEquals("tx2", registry.getHolder(repository));
                assertTrue(handler2.isGranted());
            }
        }
    }

    @Test
    public void testConcurrentRequestsInReversePriorityOrder() {
        GrantHandler handler1 = new GrantHandler();
        GrantHandler handler2 = new GrantHandler();
        GrantHandler handler3 = new GrantHandler();
        try (WriteToken.Promise ignored1 = registry.acquireToken(repository, "tx1", 1000L, handler1)) {
            //Token should be held by tx1
            assertEquals("tx1", registry.getHolder(repository));
            assertTrue(handler1.isGranted());

            //Use an earlier timestamp to simulate a higher priority request
            try (WriteToken.Promise ignored2 = registry.acquireToken(repository, "tx2", 999L, handler2)) {
                //Token should still be held by tx1, but tx1 should have been requested to yield
                assertEquals("tx1", registry.getHolder(repository));
                assertTrue(handler1.isYieldRequested());

                //Simulate tx1 yielding the token without an exception
                handler1.doYield();

                //This should release the token, which should immediately be granted to tx2
                assertEquals("tx2", registry.getHolder(repository));
                assertTrue(handler2.isGranted());
                //Token should no longer be held by tx1
                assertFalse(handler1.isGranted());

                //Make matters worse, and introduce an even higher priority request
                try (WriteToken.Promise ignored3 = registry.acquireToken(repository, "tx3", 998L, handler3)) {
                    //Token should still be held by tx2, but it too should have been requested to yield
                    assertEquals("tx2", registry.getHolder(repository));
                    assertTrue(handler2.isYieldRequested());

                    //Simulate tx2 yielding the token
                    handler2.doYield();

                    //This should release the token, which should immediately be granted to tx3
                    assertEquals("tx3", registry.getHolder(repository));
                }

                //tx3 has now released the token, which should be re-granted to tx2
                assertEquals("tx2", registry.getHolder(repository));
                assertTrue(handler2.isGranted());
            }

            //After tx2 releases the token, tx1 should be re-granted the token
            assertEquals("tx1", registry.getHolder(repository));
            assertTrue(handler1.isGranted());
        }
    }

    @Test
    public void testMaybeRetryTokenYieldRequests() throws InterruptedException {
        //Verify that a periodic retry task is scheduled
        ArgumentCaptor<Runnable> taskCaptor = ArgumentCaptor.forClass(Runnable.class);
        verify(taskScheduler).scheduleAtFixedRate(taskCaptor.capture(), eq(Duration.ofMillis(25L))); //25ms for the test
        Runnable scheduledTask = taskCaptor.getValue();

        GrantHandler handler1 = new GrantHandler();
        GrantHandler handler2 = new GrantHandler();
        try (WriteToken.Promise ignored1 = registry.acquireToken(repository, "tx1", 1000L, handler1)) {
            //Token should be held by tx1
            assertEquals("tx1", registry.getHolder(repository));
            assertTrue(handler1.isGranted());

            //Use an earlier timestamp to simulate a higher priority request
            try (WriteToken.Promise ignored2 = registry.acquireToken(repository, "tx2", 999L, handler2)) {
                //Token should still be held by tx1, but tx1 should have been requested to yield
                assertEquals("tx1", registry.getHolder(repository));
                assertTrue(handler1.isYieldRequested());

                assertEquals(1, handler1.getYieldRequestCount());

                //Running the periodic task should only trigger a retry if it's been more than 25ms (configurable)
                //since the previous request.
                scheduledTask.run();
                assertEquals(1, handler1.getYieldRequestCount());
                scheduledTask.run();
                assertEquals(1, handler1.getYieldRequestCount());

                Thread.sleep(30L);

                scheduledTask.run();
                scheduledTask.run();
                scheduledTask.run();
                assertEquals(2, handler1.getYieldRequestCount());

                //Simulate tx1 yielding the token without an exception
                handler1.doYield();

                //This should release the token, which should immediately be granted to tx2
                assertEquals("tx2", registry.getHolder(repository));
                assertTrue(handler2.isGranted());
                //Token should no longer be held by tx1
                assertFalse(handler1.isGranted());
            }
        }
    }

    @Test
    public void testRequestsFromDifferentRepositoriesDoNotCompete() {
        Repository repository2 = mock(Repository.class);
        when(repository2.getSlug()).thenReturn("repository-2");

        GrantHandler handler1 = new GrantHandler();
        GrantHandler handler2 = new GrantHandler();
        try (WriteToken.Promise ignored1 = registry.acquireToken(repository, "tx1", 1000L, handler1)) {
            assertEquals("tx1", registry.getHolder(repository));
            assertNull(registry.getHolder(repository2));

            try (WriteToken.Promise ignored2 = registry.acquireToken(repository2, "tx2", 999L, handler2)) {
                assertEquals("tx1", registry.getHolder(repository));
                assertEquals("tx2", registry.getHolder(repository2));
            }

            assertEquals("tx1", registry.getHolder(repository));
            assertNull(registry.getHolder(repository2));
        }
        assertNull(registry.getHolder(repository));
        assertNull(registry.getHolder(repository2));
    }

    @Test
    public void testSequentialRequests() {
        GrantHandler handler = new GrantHandler();
        try (WriteToken.Promise handle = registry.acquireToken(repository, "tx1", 1000L, handler)) {
            assertNotNull(handle);

            assertEquals("tx1", registry.getHolder(repository));
            assertTrue(handler.isGranted());
        }

        //The token should now be released and available
        assertNull(registry.getHolder(repository));

        handler = new GrantHandler();
        try (WriteToken.Promise handle = registry.acquireToken(repository, "tx2", 1001L, handler)) {
            assertNotNull(handle);

            assertEquals("tx2", registry.getHolder(repository));
            assertTrue(handler.isGranted());
        }
        assertNull(registry.getHolder(repository));
    }

    @Test
    public void testStartLockRequestsAndReleasesLocalToken() {
        //This test only tests the interaction with the registry. See DefaultWriteLockTest more detailed
        //testing of acquiring the lock, etc.
        DefaultWriteLock writeLock = registry.startLock(repository, "tx1", ImmutableList.of(localNode, remoteNode));

        assertNotNull(writeLock);
        assertEquals(TIMESTAMP, writeLock.getLogicalTimestamp());
        assertEquals(1, writeLock.getTokenCount());
        assertEquals("tx1", registry.getHolder(repository));

        //Trigger the token to be yielded to tx2
        WriteTokenHandler handler = mock(WriteTokenHandler.class);
        try (WriteToken.Promise ignored = registry.acquireToken(repository, "tx2", TIMESTAMP - 1L, handler)) {
            assertEquals(0, writeLock.getTokenCount());
        }

        //After the token has been released by tx2, it should be re-granted to the lock
        assertEquals(1, writeLock.getTokenCount());

        //Closing the lock should release the token and close the Promise, which retracts the request
        writeLock.close();

        assertNull(registry.getHolder(repository));
    }

    @Test
    public void testTokenIsReleasedIfGrantedCallbackThrows() {
        WriteTokenHandler handler = mock(WriteTokenHandler.class);
        doThrow(new IllegalStateException("Mock exception")).when(handler).onGranted(any());

        assertThrows(IllegalStateException.class, () -> registry.acquireToken(repository, "tx1", 1000L, handler));
        assertNull(registry.getHolder(repository));

        logListener.assertError("releasing token");
    }

    @Test
    public void testTokenIsReleasedIfGrantedCallbackThrowsOnLateGrant() {
        WriteTokenHandler handler1 = mock(WriteTokenHandler.class);
        WriteTokenHandler handler2 = mock(WriteTokenHandler.class);
        WriteTokenHandler handler3 = mock(WriteTokenHandler.class);
        doThrow(new IllegalStateException("Mock exception")).when(handler2).onGranted(any());

        //Queue up three requests
        WriteToken.Promise promise1 = registry.acquireToken(repository, "tx1", 1000L, handler1);
        WriteToken.Promise promise2 = registry.acquireToken(repository, "tx2", 1001L, handler2);
        WriteToken.Promise promise3 = registry.acquireToken(repository, "tx3", 1002L, handler3);

        //Token should be granted to handle1 now
        verify(handler1).onGranted(any());

        //Closing handle1 should grant the token to tx2, and trigger the exception. In this case, the exception
        //should be logged but not bubble up.
        promise1.close();
        verify(handler2).onGranted(any());

        logListener.assertError("releasing token");

        //The token should now be granted to tx3
        assertEquals("tx3", registry.getHolder(repository));

        //Closing handle2 should change nothing
        promise2.close();
        assertEquals("tx3", registry.getHolder(repository));

        //Close handle3 to release the token
        promise3.close();
        assertNull(registry.getHolder(repository));
    }

    private static class GrantHandler implements WriteTokenHandler {

        private final AtomicInteger yieldRequestCount = new AtomicInteger(0);

        private volatile WriteToken token;

        @Override
        public void onGranted(@Nonnull WriteToken writeToken) {
            assertNull("Token was granted while already held!", token);
            token = writeToken;
        }

        @Override
        public void yieldToken() {
            //Don't immediately yield the token for testing purposes
            yieldRequestCount.incrementAndGet();
        }

        void doYield() {
            int count = yieldRequestCount.getAndSet(0);
            assertTrue(count > 0);
            token.yieldToken();
            token = null;
        }

        int getYieldRequestCount() {
            return yieldRequestCount.get();
        }

        boolean isGranted() {
            return token != null;
        }

        boolean isYieldRequested() {
            return yieldRequestCount.get() > 0;
        }
    }
}