package com.atlassian.stash.internal.notification.batch;

import com.atlassian.sal.api.transaction.TransactionTemplate;
import com.atlassian.scheduler.JobRunner;
import com.atlassian.scheduler.SchedulerService;
import com.atlassian.scheduler.config.JobConfig;
import com.atlassian.scheduler.config.JobId;
import com.atlassian.scheduler.config.JobRunnerKey;
import com.atlassian.stash.internal.notification.batch.dao.UserNotification;
import com.atlassian.stash.internal.notification.batch.dao.UserNotificationBuilder;
import com.atlassian.stash.internal.notification.batch.dao.UserNotificationDao;
import com.atlassian.bitbucket.transaction.SimpleSalTransactionTemplate;
import com.atlassian.bitbucket.user.ApplicationUser;
import com.atlassian.bitbucket.user.UserService;
import com.atlassian.bitbucket.util.Chainable;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableListMultimap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Multimap;
import org.hamcrest.Matchers;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.Spy;
import org.mockito.runners.MockitoJUnitRunner;

import javax.annotation.Nonnull;
import java.util.*;

import static org.hamcrest.CoreMatchers.hasItems;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.*;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyLong;
import static org.mockito.Mockito.*;

@RunWith(MockitoJUnitRunner.class)
public class BatchNotificationJobTest {

    private static final ApplicationUser USER1 = user(1, "user1");
    private static final ApplicationUser USER2 = user(2, "user2");

    private TestBatchSender batchSender1 = batchSender("1");
    private TestBatchSender batchSender2 = batchSender("2");

    private int notificationId;

    @Mock
    private BatchTiming batchTiming;

    @Spy
    private List<BatchSender> batchSenders =
            new ArrayList<>(Arrays.asList(batchSender1, batchSender2));

    @Mock
    private SchedulerService schedulerService;

    @Spy
    private TransactionTemplate transactionTemplate = new SimpleSalTransactionTemplate();

    @Mock
    private UserNotificationDao userNotificationDao;

    @Mock
    private UserService userService;

    @InjectMocks
    private BatchNotificationJob service;

    @Before
    public void setup() {
        mockShouldBeSent(true);
        when(userService.getUserById(USER1.getId())).thenReturn(USER1);
        when(userService.getUserById(USER2.getId())).thenReturn(USER2);
    }

    @Test
    public void testScheduling() throws Exception {
        service.onStart();
        verify(schedulerService).registerJobRunner(any(JobRunnerKey.class), any(JobRunner.class));
        verify(schedulerService).scheduleJob(any(JobId.class), any(JobConfig.class));
    }

    @Test
    public void testNoNotifications() {
        mockNotifications();
        service.process();
        verifyZeroInteractions(userService);
    }

    @Test
    public void testShouldNotBeSent() {
        UserNotification notification1 = notification(USER1.getId(), 1000, batchSender1, "batch1", "data");
        mockNotifications(notification1);
        when(batchTiming.batchShouldBeSent(argThat(hasItems(notification1)), anyLong())).thenReturn(false);
        service.process();
        verifyZeroInteractions(userService);
        verifyNoSend(batchSender1);
        verifyNoSend(batchSender2);
    }

    @Test
    public void testSend() {
        UserNotification notification1 = notification(USER1.getId(), 1000, batchSender1, "batch1", "data");
        mockNotifications(notification1);
        service.process();
        verifySent(batchSender1, "batch1", ImmutableListMultimap.of(USER1, notification1));
        verifyDeleted(notification1);
    }

    @Test
    public void testSenderSelection() {
        UserNotification notification1 = notification(USER1.getId(), 1000, batchSender1, "batch1", "data");
        UserNotification notification2 = notification(USER1.getId(), 3000, batchSender2, "batch1", "data");
        mockNotifications(notification1, notification2);
        service.process();
        verifySent(batchSender1, "batch1", ImmutableListMultimap.of(USER1, notification1));
        verifySent(batchSender2, "batch1", ImmutableListMultimap.of(USER1, notification2));
        verifyDeleted(notification1, notification2);
    }

    @Test
    public void testGroupingByBatch() {
        UserNotification notification1 = notification(USER1.getId(), 1000, batchSender1, "batch1", "data");
        UserNotification notification2 = notification(USER1.getId(), 2000, batchSender1, "batch1", "data");
        UserNotification notification3 = notification(USER1.getId(), 3000, batchSender1, "batch2", "data");
        mockNotifications(notification1, notification2, notification3);
        service.process();
        verifySent(batchSender1, "batch1", ImmutableListMultimap.of(USER1, notification1, USER1, notification2));
        verifySent(batchSender1, "batch2", ImmutableListMultimap.of(USER1, notification3));
        verifyDeleted(notification1, notification2, notification3);
    }

    @Test
    public void testGroupingByUser() {
        UserNotification notification1 = notification(USER1.getId(), 1000, batchSender1, "batch1", "data");
        UserNotification notification2 = notification(USER2.getId(), 2000, batchSender1, "batch1", "data");
        mockNotifications(notification1, notification2);
        service.process();
        verifySent(batchSender1, "batch1", ImmutableListMultimap.of(USER1, notification1, USER2, notification2));
        verifyDeleted(notification1, notification2);
    }

    @Test
    public void testNoDeletionOnError() {
        UserNotification notification1 = notification(USER1.getId(), 1000, batchSender1, "batch1", "data");
        mockNotifications(notification1);
        batchSender1.setThrowOnSend(new RuntimeException("error in sender"));
        service.process();
        verifySent(batchSender1, "batch1", ImmutableListMultimap.of(USER1, notification1));
        verify(userNotificationDao, never()).delete(any());
    }

    @Test
    public void testPartialDeletionOnPartialSend() {
        UserNotification notification1 = notification(USER1.getId(), 1000, batchSender1, "batch1", "data1");
        UserNotification notification2 = notification(USER2.getId(), 2000, batchSender1, "batch1", "data2");
        mockNotifications(notification1, notification2);
        batchSender1.setNotificationsToMarkAsSent(ImmutableList.of(notification1));
        service.process();
        verifyDeleted(notification1);
    }

    @Test
    public void testDeletionOnErrorAfterTimeout() {
        UserNotification notification1 = notification(USER1.getId(), 1000, batchSender1, "batch1", "data");
        mockNotifications(notification1);
        batchSender1.setThrowOnSend(new RuntimeException("error in sender"));
        when(batchTiming.batchCanBeDiscarded(eq(ImmutableSet.of(notification1)), anyLong())).thenReturn(true);
        service.process();
        verifySent(batchSender1, "batch1", ImmutableListMultimap.of(USER1, notification1));
        verifyDeleted(notification1);
    }

    @Test
    public void testUserNotFound() {
        ApplicationUser user404 = user(404, "user404");
        when(userService.getUserById(user404.getId())).thenReturn(null);
        UserNotification notification1 = notification(user404.getId(), 1000, batchSender1, "batch1", "data");
        mockNotifications(notification1);
        service.process();
        verifyNoSend(batchSender1);
    }

    @Test
    public void testDiscardingWhenBatchSenderNotFound() {
        UserNotification notification1 = notification(USER1.getId(), 1000, batchSender("notregistered"), "batch1", "data");
        mockNotifications(notification1);
        service.process();
        verifyDeleted(notification1);
    }

    @Test
    public void testSendersWithDuplicateIds() {
        String id = "duplicateId";
        TestBatchSender sender1 = batchSender(id);
        TestBatchSender sender2 = batchSender(id);

        List<BatchSender> duplicateSenders = ImmutableList.<BatchSender>of(sender1, sender2);
        BatchNotificationJob service = new BatchNotificationJob(duplicateSenders, batchTiming, userNotificationDao,
                userService, schedulerService, transactionTemplate);

        UserNotification notification1 = notification(USER1.getId(), 1000, sender1, "batch1", "data");
        mockNotifications(notification1);

        service.process();

        verifySent(sender1, "batch1", ImmutableListMultimap.of(USER1, notification1));
    }

    @Test
    public void testGetCount() throws Exception {
        when(userNotificationDao.count()).thenReturn(5);
        assertThat(service.notificationCount(), is(5));
    }

    private void mockShouldBeSent(boolean shouldBeSent) {
        when(batchTiming.batchShouldBeSent(any(), anyLong())).thenReturn(shouldBeSent);
    }

    private void mockNotifications(UserNotification... notifications) {
        when(userNotificationDao.findAll()).thenReturn(Arrays.asList(notifications));
    }

    private static ApplicationUser user(int id, String name) {
        ApplicationUser user = mock(ApplicationUser.class);
        when(user.getId()).thenReturn(id);
        when(user.getName()).thenReturn(name);
        return user;
    }

    private static TestBatchSender batchSender(String id) {
        return new TestBatchSender(id);
    }

    private void verifySent(TestBatchSender batchSender, String batchId, Multimap<ApplicationUser, UserNotification> notifications) {
        assertThat(batchSender.getActualNotifications(batchId), is(notifications));
    }

    private void verifyNoSend(TestBatchSender batchSender) {
        assertThat(batchSender.getActualNotifications().size(), is(0));
    }

    private void verifyDeleted(UserNotification... notifications) {
        List<Long> ids = Chainable.chain(notifications).transform(UserNotification::getId).toList();
        verify(userNotificationDao).delete((Iterable<Long>) (argThat(Matchers.contains(ids.toArray()))));
    }

    private UserNotification notification(int userId, long timestamp, BatchSender batchSender, String batchId, String data) {
        return new UserNotificationBuilder()
                .id(notificationId++)
                .userId(userId)
                .timestamp(timestamp)
                .batchSenderId(batchSender.getId())
                .batchId(batchId)
                .data(data)
                .build();
    }

    private static class TestBatchSender implements BatchSender {

        private final String id;
        private RuntimeException throwOnSend;

        private Map<String, Multimap<ApplicationUser, UserNotification>> actualNotifications = new HashMap<>();
        private Collection<UserNotification> notificationsToMarkAsSent;

        public TestBatchSender(String id) {
            this.id = id;
        }

        public void setThrowOnSend(RuntimeException e) {
            this.throwOnSend = e;
        }

        public void setNotificationsToMarkAsSent(Collection<UserNotification> notificationsToMarkAsSent) {
            this.notificationsToMarkAsSent = notificationsToMarkAsSent;
        }

        public Map<String, Multimap<ApplicationUser, UserNotification>> getActualNotifications() {
            return actualNotifications;
        }

        public Multimap<ApplicationUser, UserNotification> getActualNotifications(String batchId) {
            return actualNotifications.get(batchId);
        }

        @Nonnull
        @Override
        public String getId() {
            return id;
        }

        @Override
        public void send(@Nonnull String batchId, @Nonnull Multimap<ApplicationUser, UserNotification> notifications,
                @Nonnull Collection<UserNotification> sentNotifications) {
            if (actualNotifications.containsKey(batchId)) {
                fail("Multiple calls for the same batch: " + batchId);
            }
            actualNotifications.put(batchId, notifications);
            if (throwOnSend != null) {
                throw throwOnSend;
            }

            if (notificationsToMarkAsSent != null) {
                sentNotifications.addAll(notificationsToMarkAsSent);
            } else {
                sentNotifications.addAll(notifications.values());
            }
        }
    }
}
