package com.atlassian.stash.internal.idx;

import com.atlassian.bitbucket.scm.*;
import com.atlassian.plugin.ModuleDescriptor;
import com.atlassian.plugin.Plugin;
import com.atlassian.plugin.event.events.PluginDisabledEvent;
import com.atlassian.plugin.event.events.PluginEnabledEvent;
import com.atlassian.bitbucket.commit.Commit;
import com.atlassian.bitbucket.commit.CommitCallback;
import com.atlassian.bitbucket.commit.CommitContext;
import com.atlassian.bitbucket.commit.CommitSummary;
import com.atlassian.bitbucket.NoSuchResourceException;
import com.atlassian.bitbucket.i18n.I18nService;
import com.atlassian.bitbucket.i18n.KeyedMessage;
import com.atlassian.bitbucket.idx.CommitIndex;
import com.atlassian.bitbucket.idx.CommitIndexer;
import com.atlassian.bitbucket.idx.CommitIndexingException;
import com.atlassian.bitbucket.idx.IndexingContext;
import com.atlassian.stash.internal.project.InternalProject;
import com.atlassian.stash.internal.repository.InternalRepository;
import com.atlassian.bitbucket.repository.Repository;
import com.atlassian.bitbucket.repository.RepositoryService;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import org.hamcrest.BaseMatcher;
import org.hamcrest.Description;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentCaptor;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.internal.matchers.GreaterThan;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.runners.MockitoJUnitRunner;
import org.mockito.stubbing.Answer;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.PropertyAccessor;
import org.springframework.transaction.PlatformTransactionManager;

import java.util.Collections;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

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

@RunWith(MockitoJUnitRunner.class)
@SuppressWarnings("unchecked")
public class DefaultCommitIndexingServiceTest {

    public static final String PREVIOUS = "previous";
    public static final String CURRENT = "current";
    public static final Iterable<String> LAST_HEADS = Collections.singleton(PREVIOUS);
    public static final Iterable<String> CURRENT_HEADS = Collections.singleton(CURRENT);

    @Mock
    private CommitIndex commitIndex;
    @Mock
    private ScmCommandFactory commandFactory;
    @Mock
    private I18nService i18nService;
    @Mock
    private IndexerStateDao indexerStateDao;
    @Mock
    private RepositoryService repositoryService;
    @Mock
    private ScmService scmService;
    @InjectMocks
    private DefaultCommitIndexingService service;
    @Mock
    private RepositorySnapshotService snapshotService;
    @Mock
    private PlatformTransactionManager transactionManager;

    @Before
    public void setup() throws Exception {
        service.setIndexBatchSize(2);
        service.setQueueSize(1024);

        when(i18nService.createKeyedMessage(anyString(), anyVararg())).thenReturn(new KeyedMessage("", "", ""));
        when(scmService.getCommandFactory(any(Repository.class))).thenReturn(commandFactory);
        when(indexerStateDao.getOldestLastRunTimestamp(any(InternalRepository.class), any(Iterable.class))).thenReturn(100L);
        when(snapshotService.getByRepository(any(Repository.class), eq(100L))).thenReturn(LAST_HEADS);
        when(snapshotService.getByRepository(any(Repository.class), longThat(new GreaterThan<>(100L)))).thenReturn(CURRENT_HEADS);
        when(snapshotService.create(any(Repository.class), longThat(new GreaterThan<>(100L)))).thenReturn(CURRENT_HEADS);
    }

    @Test(expected = CommitIndexingException.class)
    public void testAbortOnTimeout() throws Exception {
        // set the timeout to 1s
        PropertyAccessor accessor = new DirectFieldAccessor(service);
        accessor.setPropertyValue("timeoutSeconds", "1");

        Repository repository = mockRepo("repo", 1, "project");

        // register a sample indexer
        mockIndexer(repository, true);

        // mock the scm operation to retrieve the descendant commits
        IndexingCommand command = mock(IndexingCommand.class);
        when(command.asynchronous()).thenReturn(command);
        final ArgumentCaptor<CommitCallback> callbackCaptor = ArgumentCaptor.forClass(CommitCallback.class);
        when(commandFactory.commits(any(CommitsCommandParameters.class), callbackCaptor.capture())).thenReturn(command);

        final Future<Void> future = mock(Future.class);
        // simulate that the process is still running
        when(future.get(anyLong(), any(TimeUnit.class))).thenThrow(new TimeoutException());

        when(command.start()).thenAnswer(new Answer<Future<Void>>() {
            @Override
            public Future<Void> answer(InvocationOnMock invocation) throws Throwable {
                new Thread(new Runnable() {
                    @Override
                    public void run() {
                        CommitCallback callback = callbackCaptor.getValue();
                        try {
                            callback.onStart(mock(CommitContext.class));
                            Thread.sleep(1050L);
                            callback.onCommit(mock(Commit.class));
                            callback.onEnd(mock(CommitSummary.class));
                        } catch (Exception e) {
                            throw new RuntimeException(e);
                        }
                    }
                }).start();
                return future;
            }
        });

        service.indexRepository(repository);
    }

    @Test
    public void testPluginEnabledDisabledDetection() {
        assertTrue(service.getIndexers().isEmpty());

        // create a mock plugin and a mock PluginEnabledEvent
        CommitIndexer indexer = mock(CommitIndexer.class);
        when(indexer.getId()).thenReturn("mockIndexer");
        ModuleDescriptor<CommitIndexer> module = mock(ModuleDescriptor.class);
        PluginEnabledEvent pluginEnabledEvent = mock(PluginEnabledEvent.class);
        Plugin plugin = mock(Plugin.class);

        when(pluginEnabledEvent.getPlugin()).thenReturn(plugin);
        when(plugin.getKey()).thenReturn("key");
        when(plugin.getModuleDescriptorsByModuleClass(eq(CommitIndexer.class))).thenReturn(Lists.newArrayList(module));
        when(module.getModule()).thenReturn(indexer);

        service.onPluginEnabled(pluginEnabledEvent);

        // verify that the plugin has been registered
        assertEquals(1, service.getIndexers().size());
        assertSame(indexer, service.getIndexers().iterator().next());

        // now fake a plugin disabled event. For some reason, the ModuleDescriptors are no longer available when the
        // PluginDisabledEvent is raised.

        reset(plugin);
        PluginDisabledEvent disabledEvent = mock(PluginDisabledEvent.class);
        when(disabledEvent.getPlugin()).thenReturn(plugin);
        when(plugin.getKey()).thenReturn("key");

        service.onPluginDisabled(disabledEvent);

        // verify that the indexer has unregistered
        assertTrue(service.getIndexers().isEmpty());
    }

    @Test
    public void testFailingIndexerDoesNotAffectOtherIndexers() {
        Repository repository = mockRepo("repo", 1, "project");

        // register a sample indexer
        CommitIndexer indexer = mockIndexer(repository, true);
        CommitIndexer bogusIndexer = mockIndexer(repository, true);

        // mock the scm interactions
        mockGetCommitsBetweenCommand(CURRENT_HEADS, LAST_HEADS, 3, null);
        mockGetCommitsBetweenCommand(LAST_HEADS, CURRENT_HEADS, 2, null);

        // mock the failing indexer
        doThrow(new NullPointerException())
                .when(bogusIndexer).onCommitAdded(any(Commit.class), any(IndexingContext.class));

        service.indexRepository(repository);

        // verify the interactions with the CommitIndex
        // expecting three added and two deleted commits
        verify(commitIndex, times(3)).addCommit(any(Commit.class), same(repository));
        verify(commitIndex, times(2)).removeCommit(anyString(), same(repository));

        // verify the interactions on the commit indexer
        verify(indexer, times(1)).onBeforeIndexing(any(IndexingContext.class));
        verify(indexer, times(3)).onCommitAdded(any(Commit.class), any(IndexingContext.class));
        verify(indexer, times(2)).onCommitRemoved(any(Commit.class), any(IndexingContext.class));
        verify(indexer, times(1)).onAfterIndexing(any(IndexingContext.class));

        // verify the interactions on the bogusIndexer
        verify(bogusIndexer, times(1)).onBeforeIndexing(any(IndexingContext.class));
        // exception thrown on the first
        verify(bogusIndexer, times(1)).onCommitAdded(any(Commit.class), any(IndexingContext.class));
        verify(bogusIndexer, never()).onCommitRemoved(any(Commit.class), any(IndexingContext.class));
        verify(bogusIndexer, times(1)).onAfterIndexing(any(IndexingContext.class));
    }

    @Test
    public void testIndexRepository() throws Exception {
        Repository repository = mockRepo("repo", 1, "project");

        // register a sample indexer
        CommitIndexer indexer = mockIndexer(repository, true);

        // mock the scm interactions
        mockGetCommitsBetweenCommand(CURRENT_HEADS, LAST_HEADS, 3, null);
        mockGetCommitsBetweenCommand(LAST_HEADS, CURRENT_HEADS, 2, null);

        service.indexRepository(repository);

        // verify the interactions with the CommitIndex
        // expecting three added and two deleted commits
        verify(commitIndex, times(3)).addCommit(any(Commit.class), same(repository));
        verify(commitIndex, times(2)).removeCommit(anyString(), same(repository));

        // verify the interactions on the commit indexer
        verify(indexer, times(1)).onBeforeIndexing(any(IndexingContext.class));
        verify(indexer, times(3)).onCommitAdded(any(Commit.class), any(IndexingContext.class));
        verify(indexer, times(2)).onCommitRemoved(any(Commit.class), any(IndexingContext.class));
        verify(indexer, times(1)).onAfterIndexing(any(IndexingContext.class));
    }

    @Test
    public void testIndexRepositoryWithoutIndexers() throws Exception {
        Repository repository = mockRepo("repo", 1, "project");
        when(repositoryService.getById(1)).thenReturn(repository);

        service.indexRepository(repository);

        // verify that no indexing has been triggered
        verifyZeroInteractions(transactionManager, indexerStateDao, scmService, snapshotService);
    }

    @Test
    public void testIndexRepositoryWithMissingSnaphotRescansFullHistory() {
        InternalRepository repository = mockRepo("repo", 1, "project");

        // register a sample indexer
        CommitIndexer indexer = mockIndexer(repository, true);

        when(indexerStateDao.getOldestLastRunTimestamp(same(repository), any(Iterable.class))).thenReturn(12345L);
        NoSuchResourceException exception = new NoSuchResourceException(mock(KeyedMessage.class), "");
        when(snapshotService.getByRepository(same(repository), eq(12345L)))
                .thenThrow(exception);

        // mock the scm interactions
        Iterable<String> emptyHeads = Collections.emptySet();
        mockGetCommitsBetweenCommand(CURRENT_HEADS, emptyHeads, 3, null);
        mockGetCommitsBetweenCommand(emptyHeads, CURRENT_HEADS, 2, null);


        service.indexRepository(repository);

        // verify the interactions with the CommitIndex
        // expecting three added and two deleted commits
        verify(commitIndex, times(3)).addCommit(any(Commit.class), same(repository));
        verify(commitIndex, times(2)).removeCommit(anyString(), same(repository));

        // verify the interactions on the commit indexer
        verify(indexer, times(1)).onBeforeIndexing(any(IndexingContext.class));
        verify(indexer, times(3)).onCommitAdded(any(Commit.class), any(IndexingContext.class));
        verify(indexer, times(2)).onCommitRemoved(any(Commit.class), any(IndexingContext.class));
        verify(indexer, times(1)).onAfterIndexing(any(IndexingContext.class));
    }

    private Future<Void> mockGetCommitsBetweenCommand(Iterable<String> includedHeads, Iterable<String> excludedHeads,
            final int commitCount, final IndexingControl control) {
        // mock the commits between command
        final IndexingCommand command = mock(IndexingCommand.class);
        when(command.asynchronous()).thenReturn(command);
        when(commandFactory.commits(argThat(new CommitsBetweenMatcher(includedHeads, excludedHeads)),
                any(CommitCallback.class))).thenAnswer(
                new Answer<AsyncCommand<Void>>() {
                    @Override
                    public AsyncCommand<Void> answer(InvocationOnMock invocation) throws Throwable {
                        // mock commitCount removed commits
                        CommitCallback callback = (CommitCallback) invocation.getArguments()[1];
                        callback.onStart(mock(CommitContext.class));
                        for (int i = 0; i < commitCount; ++i) {
                            callback.onCommit(mock(Commit.class));
                        }

                        if (control != null) {
                            // to allow testing for Thread interactions
                            control.onBeforeEnd();
                        }
                        callback.onEnd(mock(CommitSummary.class));

                        return command;
                    }
                });
        Future<Void> future = mock(Future.class);
        when(command.start()).thenReturn(future);

        return future;
    }

    private CommitIndexer mockIndexer(Repository repository, boolean enabled) {
        CommitIndexer indexer = mock(CommitIndexer.class);
        when(indexer.isEnabledForRepository(same(repository))).thenReturn(enabled);
        when(indexer.getId()).thenReturn("mockIndexer");
        service.register(indexer);

        return indexer;
    }

    private InternalRepository mockRepo(String repoSlug, int repoId, String projectKey) {
        InternalRepository repo = mock(InternalRepository.class);
        when(repo.getId()).thenReturn(repoId);
        when(repo.getSlug()).thenReturn(repoSlug);

        InternalProject project = mock(InternalProject.class);
        when(project.getKey()).thenReturn(projectKey);
        when(repo.getProject()).thenReturn(project);

        when(repositoryService.getById(1)).thenReturn(repo);

        return repo;
    }

    private interface IndexingCommand extends Command<Void>, AsyncCommand<Void> {
    }

    private interface IndexingControl {
        void onBeforeEnd();
    }

    private static class CommitsBetweenMatcher extends BaseMatcher<CommitsCommandParameters> {
        private Iterable<String> includes;
        private Iterable<String> excludes;

        public CommitsBetweenMatcher(Iterable<String> includes, Iterable<String> excludes) {
            this.includes = includes;
            this.excludes = excludes;
        }

        @Override
        public boolean matches(Object o) {
            if (!(o instanceof CommitsCommandParameters)) {
                return false;
            }

            CommitsCommandParameters parameters = (CommitsCommandParameters) o;
            return Iterables.elementsEqual(parameters.getIncludes(), includes) &&
                    Iterables.elementsEqual(parameters.getExcludes(), excludes);
        }

        @Override
        public void describeTo(Description description) {
        }
    }
}
