package com.atlassian.bitbucket.mesh.throttle;

import com.atlassian.bitbucket.mesh.execution.ExecutionContext;
import com.atlassian.bitbucket.mesh.execution.GrpcExecutionManager;
import com.atlassian.bitbucket.mesh.junit.LogListener;
import com.atlassian.bitbucket.mesh.test.ExecutionContextRule;
import com.google.common.collect.ImmutableMap;
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.Spy;
import org.mockito.junit.MockitoJUnitRunner;
import org.springframework.scheduling.TaskScheduler;

import java.time.Clock;
import java.time.Duration;
import java.util.Optional;
import java.util.OptionalLong;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

import static com.atlassian.bitbucket.mesh.throttle.ThrottledResource.*;
import static org.junit.Assert.*;
import static org.mockito.Mockito.*;

@SuppressWarnings("resource")
@RunWith(MockitoJUnitRunner.class)
public class SemaphoreThrottleServiceTest {

    private static final ThrottledResource RESOURCE = LFS;
    private static final long TIME_START = 1474633660L;

    @Rule
    public ExecutionContextRule executionRule = new ExecutionContextRule(new GrpcExecutionManager());
    @Rule
    public LogListener logListener = new LogListener(SemaphoreThrottleService.class);

    @Spy
    @SuppressWarnings("MismatchedQueryAndUpdateOfCollection")
    private final Properties properties = new Properties();

    @Mock
    private Clock clock;
    private SemaphoreThrottleService service;
    @Mock
    private TaskScheduler taskScheduler;
    @Mock
    private ResourceThrottleStrategyProvider throttleStrategyProvider;

    @Before
    public void setUp() {
        when(clock.millis()).thenAnswer(invocation -> Clock.systemDefaultZone().millis());
        service = new SemaphoreThrottleService(clock, executionRule.getExecutionManager(), taskScheduler,
                throttleStrategyProvider);
    }

    @Test
    public void testAcquireTicket() {
        Ticket ticket = acquireTicket();
        assertNotNull(ticket);
        assertEquals(RESOURCE, ticket.getResource());
        assertTicketsInUse(RESOURCE, 1);

        Ticket ticket2 = service.acquireTicket(RESOURCE);
        assertNotSame(ticket, ticket2);
        assertNotNull(ticket2);
        assertEquals(RESOURCE, ticket.getResource());
        assertTicketsInUse(RESOURCE, 1);

        ticket2.close();
        assertTicketsInUse(RESOURCE, 1);

        ticket.close();
        assertTicketsInUse(RESOURCE, 0);
    }

    @Test(expected = ResourceBusyException.class)
    public void testAcquireTicketForExhaustedResourceTime() throws Exception {
        mockStrategies(RESOURCE, new FixedResourceThrottleStrategy(RESOURCE, 0, 0));

        service.initialize();
        try {
            service.acquireTicket(RESOURCE);
        } catch (ResourceBusyException e) {
            Thread.sleep(10); //we can be too fast here... i.e. time is 0 but we have rejected
            assertNotSame(0L, service.getTimeSinceLastRejectedTicketRequest(RESOURCE));

            logListener.assertWarn("A [" + RESOURCE + "] ticket could not be acquired (0/0)");
            throw e;
        }
    }

    @Test(expected = NoSuchResourceException.class)
    public void testAcquireTicketForUnconfiguredResource() {
        service.acquireTicket(RESOURCE);
    }

    @Test
    public void testAcquireTicketHandlesInterruptedException() throws InterruptedException {
        // Set the acquire timeout to 5 seconds, and only 1 ticket available
        mockStrategies(RESOURCE, new FixedResourceThrottleStrategy(RESOURCE, 5, 1));
        service.initialize();

        // Acquire a ticket here (only 1 available)
        service.acquireTicket(RESOURCE);

        // Try to acquire a ticket in a separate thread (waiting up to the acquire timeout of 5 seconds)
        Thread thread = new Thread(() -> {
            try {
                executionRule.runInNewContext(() -> service.acquireTicket(RESOURCE));
            } catch (ResourceBusyException e) {
                // Do nothing
            }
        }, getClass().getSimpleName() + ".testAcquireTicketForInterruptedException");
        thread.start();
        // Interrupt the waiting thread
        thread.interrupt();
        thread.join();

        assertSame(0L, service.getTimeSinceLastRejectedTicketRequest(RESOURCE));
    }

    @Test
    public void testAdaptiveResourcesUpdatesAccordingToSchedule() {
        AdaptiveResourceThrottleStrategy adaptive1 = mockAdaptive(20L); // update every 20s
        AdaptiveResourceThrottleStrategy throwing = mockAdaptive(30L); // update every 30s
        AdaptiveResourceThrottleStrategy adaptive2 = mockAdaptive(30L); // update every 30s
        doThrow(RuntimeException.class).when(throwing).update(any()); // to test exception handling

        FixedResourceThrottleStrategy fixed = mockFixed(MIRROR_PACK);
        mockStrategies(PACK, adaptive1, REFS, throwing, LFS, adaptive2, MIRROR_PACK, fixed);
        when(clock.millis()).thenReturn(TIME_START);

        service.initialize();

        // verify that the update job is scheduled
        ArgumentCaptor<Runnable> taskCaptor = ArgumentCaptor.forClass(Runnable.class);
        ArgumentCaptor<Duration> durationCaptor = ArgumentCaptor.forClass(Duration.class);

        verify(taskScheduler).scheduleAtFixedRate(taskCaptor.capture(), durationCaptor.capture());

        Runnable task = taskCaptor.getValue();
        Duration duration = durationCaptor.getValue();
        assertNotNull(task);
        // biggest common denominator of 20 and 30 is 10
        long intervalMillis = TimeUnit.SECONDS.toMillis(10L);
        assertEquals(intervalMillis, duration.toMillis());

        // invoke the job at regular intervals
        for (int i = 0; i <= 7; ++i) {
            when(clock.millis()).thenReturn(TIME_START + i * intervalMillis);

            task.run();

            verify(adaptive1, times(1 + i / 2)).update(any()); // should be called at i = 0, 2, 4, 6
            verify(adaptive2, times(1 + i / 3)).update(any()); // should be called at i = 0, 3, 6
            verify(fixed, never()).update(any());
        }
    }

    @Test
    public void testGetTicket() {
        assertEquals(Optional.empty(), service.getTicket());

        Ticket ticket = acquireTicket();
        assertEquals(ticket, service.getTicket().orElse(null));

        ticket.close();
        assertEquals(Optional.empty(), service.getTicket());
    }

    @Test(expected = NoSuchResourceException.class)
    public void testHasRecentlyRejectedTicketForUnconfiguredResource() {
        service.getTimeSinceLastRejectedTicketRequest(RESOURCE);
    }

    @Test
    public void testQueuingResourceTime() throws InterruptedException {
        mockStrategies(RESOURCE, new FixedResourceThrottleStrategy(RESOURCE, 2000, 1));

        service.initialize();
        Ticket ticket = service.acquireTicket(RESOURCE);
        Thread thread = new Thread(() -> executionRule.runInNewContext(() -> service.acquireTicket(RESOURCE)),
                getClass().getSimpleName() + ".testQueuingResourceTime");
        thread.start();

        //wait until we are queuing...
        long timeout = System.currentTimeMillis() + 200000L;
        do {
            Thread.sleep(100L);
        } while (System.currentTimeMillis() < timeout &&
                 service.getBucketOrThrow(RESOURCE, true).getEarliestQueuingTime() == 0L);

        assertNotEquals(0L, service.getLongestQueueingTimeForCurrentTicketRequests(RESOURCE));

        //tidy up by closing the ticket and joining the thread
        ticket.close();
        thread.join();
    }

    @Test
    public void testReleaseTicketFromOtherExecutionContext() throws Exception {
        mockStrategies(RESOURCE, new FixedResourceThrottleStrategy(RESOURCE, 0, 5));
        service.initialize();

        Ticket ticket1 = service.acquireTicket(RESOURCE);
        ExecutionContext secondContext = executionRule.newContext();
        Ticket ticket2 = secondContext.call(() -> service.acquireTicket(RESOURCE));
        assertTicketsInUse(RESOURCE, 2);

        executionRule.runInNewContext(ticket1::close);
        assertTicketsInUse(RESOURCE, 1);

        //Closing the ticket again should not result in an error and should not release the remaining ticket
        ticket1.close();
        assertTicketsInUse(RESOURCE, 1);

        ticket2.close();
        assertTicketsInUse(RESOURCE, 0);
    }

    @Test
    public void testScheduleWithOnlyFixedThrottlingDoesNotScheduleJob() {
        mockStrategies(RESOURCE, new FixedResourceThrottleStrategy(RESOURCE, 0, 0));

        service.initialize();

        verifyNoInteractions(taskScheduler);
    }

    @Test
    public void testTicketIsReleasedOnExecutionContextCleanup() {
        executionRule.runInNewContext(() -> {
            acquireTicket();
            assertTicketsInUse(RESOURCE, 1);
        });

        assertTicketsInUse(RESOURCE, 0);
    }

    private static AdaptiveResourceThrottleStrategy mockAdaptive(long intervalSec) {
        AdaptiveResourceThrottleStrategy
                strategy = mock(AdaptiveResourceThrottleStrategy.class);
        when(strategy.getUpdateInterval(any(TimeUnit.class))).thenAnswer(invocation ->
                OptionalLong.of(((TimeUnit) invocation.getArguments()[0]).convert(intervalSec, TimeUnit.SECONDS)));
        when(strategy.create()).thenReturn(new SemaphoreTicketBucket(RESOURCE, 10, 25, TimeUnit.SECONDS));
        return strategy;
    }

    private static FixedResourceThrottleStrategy mockFixed(ThrottledResource resource) {
        FixedResourceThrottleStrategy
                strategy = mock(FixedResourceThrottleStrategy.class);
        when(strategy.create()).thenReturn(new SemaphoreTicketBucket(resource, 10, 25, TimeUnit.SECONDS));
        when(strategy.getUpdateInterval(any())).thenReturn(OptionalLong.empty());
        return strategy;
    }

    private Ticket acquireTicket() {
        mockStrategies(RESOURCE, new FixedResourceThrottleStrategy(RESOURCE, 0, 1));
        service.initialize();

        return service.acquireTicket(RESOURCE);
    }

    private void assertTicketsInUse(ThrottledResource resource, int count) {
        SemaphoreTicketBucket bucket = service.getBucketOrThrow(resource, true);
        assertEquals(count, bucket.summarize().getUsed());
    }

    private void mockStrategies(Object...keysAndStrategies) {
        ImmutableMap.Builder<ThrottledResource, ResourceThrottleStrategy> builder = ImmutableMap.builder();
        for (int i = 0; i < keysAndStrategies.length; i += 2) {
            ThrottledResource resource = (ThrottledResource) keysAndStrategies[i];
            ResourceThrottleStrategy
                    strategy = (ResourceThrottleStrategy) keysAndStrategies[i + 1];
            builder.put(resource, strategy);
        }
        when(throttleStrategyProvider.getAll()).thenReturn(builder.build());
    }
}
