package com.atlassian.bitbucket.mesh.git.hook;

import com.atlassian.bitbucket.mesh.config.HooksConfig;
import com.atlassian.bitbucket.mesh.util.ConcurrencyUtils;
import com.atlassian.bitbucket.mesh.util.ExecutorUtils;
import org.junit.Test;

import java.time.Duration;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.same;
import static org.mockito.Mockito.*;

public class GitHookThreadPoolExecutorTest {

    private final ThreadFactory threadFactory = ConcurrencyUtils.createDaemonThreadFactory("git-hook-callback");

    @Test(timeout = 5000L)
    public void testRejectedExecution() throws Exception {
        CountDownLatch allComplete = new CountDownLatch(2);
        CountDownLatch inCallback = new CountDownLatch(1);
        CountDownLatch slowCallback = new CountDownLatch(1);

        GitHookThreadPoolExecutor executor = new GitHookThreadPoolExecutor(config(0, 1, 1), threadFactory);
        //Replace the "real" RejectedExecutionHandler with a spy so we can verify it's called. Note that,
        //with how the calls work, spying on the HookThreadPoolExecutor does _not_ allow detecting whether
        //its rejectedExecution method is called or not
        RejectedExecutionHandler spyHandler = spy(executor.getRejectedExecutionHandler());
        executor.setRejectedExecutionHandler(spyHandler);

        executor.execute(() -> {
            inCallback.countDown();

            try {
                slowCallback.await();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }

            allComplete.countDown();
        });
        //Wait for the "slow hook" to be called, using up the one thread the executor has. At that point,
        //whether that thread has actually blocked on the "slowCallback" latch or not, we know any other
        //submitted task will end up getting rejected
        inCallback.await();

        Thread waiter = new Thread(() -> executor.execute(allComplete::countDown), "rejected-callback");
        waiter.setDaemon(true);
        waiter.start();
        awaitRejection(waiter);

        //Once our task has been rejected, and then blocked in put, let the "slow hook" complete
        slowCallback.countDown();
        //Once the slow hook completes, the rejected task should be picked up and complete as well
        allComplete.await();

        //Verify the task was rejected at first
        verify(spyHandler).rejectedExecution(any(), same(executor));
    }

    /**
     * Verifies any thread blocked in the {@link java.util.concurrent.BlockingQueue#put blocking put} performed by
     * {@link GitHookThreadPoolExecutor#rejectedExecution} is released when the executor terminates.
     * <p>
     * This test <i>intentionally</i> doesn't verify whether the "rejected" task was eventually run or not. For the
     * purposes of this test, that's irrelevant. Instead, the focus is on ensuring the rejection handler won't delay
     * application shutdown because the blocking put never returns.
     */
    @Test(timeout = 5000L)
    public void testRejectedExecutionHandlesShutdown() throws Exception {
        CountDownLatch latch = new CountDownLatch(1);

        GitHookThreadPoolExecutor executor = new GitHookThreadPoolExecutor(config(0, 1, 1), threadFactory);
        executor.execute(() -> {
            try {
                //shutdownNow will interrupt the worker thread, which should allow this to complete
                latch.await();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        Thread waiter = new Thread(() -> executor.execute(() -> { }), "rejected-callback");
        waiter.setDaemon(true);
        waiter.start();
        awaitRejection(waiter);

        //Ideally this would call stop(), but we don't want the delay between shutdown and shutdownNow,
        //so instead we use ExecutorUtils.shutdown directly.
        //We pass 0 seconds for the shutdown timeout because, with how our "slow hook" is implemented
        //it can't possibly terminate gracefully--it can only terminate due to interruption. That means
        //any shutdown delay simply extends the test duration for no purpose.
        ExecutorUtils.shutdown(executor, 0L, TimeUnit.SECONDS, 3L, TimeUnit.SECONDS);

        //Verify that the executor was terminated, and that the waiting thread was able to complete
        assertTrue(executor.isShutdown());
        assertTrue(executor.isTerminated());
        waiter.join();
    }

    private void awaitRejection(Thread waiter) throws InterruptedException {
        long timeout = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(3L);
        while (System.currentTimeMillis() < timeout) {
            //SynchronousQueue uses LockSupport.park/parkNanos, so "waiter" should end up in WAITING state
            //rather than BLOCKED (or TIMED_WAITING since we don't use fixed timeouts)
            if (waiter.getState() == Thread.State.WAITING) {
                break;
            }

            Thread.sleep(50L); //Sleep to let the other threads "run", if they're not waiting yet
        }
    }

    private static HooksConfig config(int coreThreads, int maxThreads, int keepAliveMinutes) {
        HooksConfig config = mock(HooksConfig.class);
        when(config.getThreadsCore()).thenReturn(coreThreads);
        when(config.getThreadsMax()).thenReturn(maxThreads);
        when(config.getThreadsKeepAlive()).thenReturn(Duration.ofMinutes(keepAliveMinutes));
        return config;
    }
}
