package com.atlassian.stash.internal.scheduling;

import com.atlassian.bitbucket.annotation.Dao;
import com.atlassian.bitbucket.annotation.DataSet;
import com.atlassian.bitbucket.context.DelegatingTestExecutionListener;
import com.atlassian.scheduler.caesium.spi.ClusteredJob;
import com.atlassian.scheduler.config.JobId;
import com.atlassian.scheduler.config.JobRunnerKey;
import com.atlassian.scheduler.config.Schedule;
import com.atlassian.stash.internal.DataSets;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.TestExecutionListeners;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * This DAO test does <em>not</em> extend AbstractDaoTest because, unlike most DAO classes,
 * {@link HibernateClusteredJobDao} defines its own transaction boundaries so we do <em>not</em> need or want the
 * {@code @Transactional} annotation on any methods in this class.
 */
@ContextConfiguration(locations = "classpath:/spring/dao-test-context.xml")
@DataSet(value = {DataSets.CLUSTERED_JOBS}, delete = true)
@RunWith(SpringJUnit4ClassRunner.class)
@TestExecutionListeners(DelegatingTestExecutionListener.class)
public class HibernateClusteredJobDaoTest extends Assert {

    private static final ClusteredJob JOB1 = new InternalClusteredJob.Builder(
            JobId.of("BatchNotificationJobRunner"),
            JobRunnerKey.of("com.atlassian.stash.internal.notification.batch.BatchNotificationJob$BatchNotificationJobRunner"),
            Schedule.forInterval(60000L, new Date()))
            .build();
    private static final ClusteredJob JOB2 = new InternalClusteredJob.Builder(JOB1).build();
    private static final ClusteredJob JOB3 = new InternalClusteredJob.Builder(
            JobId.of("CleanupEmptyRescopesJob"),
            JobRunnerKey.of("com.atlassian.stash.internal.pull.rescope.DefaultRescopeProcessor$CleanupEmptyRescopesJob"),
            Schedule.forInterval(1800000L, new Date()))
            .build();
    private static final ClusteredJob JOB4 = new InternalClusteredJob.Builder(
            JobId.of("SchedulerTest.cron"),
            JobRunnerKey.of("SchedulerTest.cron"),
            Schedule.forCronExpression("0 0/1 * * * ?"))
            .build();
    private static final ClusteredJob JOB5 = new InternalClusteredJob.Builder(
            JobId.of("CompatibilityPluginScheduler.JobId.LocalPluginLicenseNotificationJob-job"),
            JobRunnerKey.of("CompatibilityPluginScheduler.JobRunnerKey.LocalPluginLicenseNotificationJob-job-handler"),
            Schedule.forCronExpression("0/5 14,18,3-39,52 * ? JAN,MAR,SEP MON-FRI 2016-2021", TimeZone.getDefault()))
            .build();

    private final Date NEXT_RUN_DATE;

    @Dao
    private HibernateClusteredJobDao dao;

    public HibernateClusteredJobDaoTest() throws ParseException {
        NEXT_RUN_DATE = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").parse("2016-03-09 11:03:01.897");
    }

    @Test
    public void testCreate() {
        try {
            assertTrue(dao.create(JOB1));
            assertEquals(JOB1, dao.find(JOB1.getJobId()));
            assertFalse(dao.create(JOB1));  // can't have duplicate primary key "BatchNotificationJobRunner"
            assertFalse(dao.create(JOB2));  // can't have duplicate primary key "BatchNotificationJobRunner"
        } finally {
            dao.delete(JOB1.getJobId());
        }
        assertFalse(dao.create(JOB3));      // can't have duplicate primary key "CleanupEmptyRescopesJob"
        try {
            assertTrue(dao.create(JOB4));
            assertEquals(JOB4, dao.find(JOB4.getJobId()));
        } finally {
            dao.delete(JOB4.getJobId());
        }
        try {
            assertTrue(dao.create(JOB5));
            assertEquals(JOB5, dao.find(JOB5.getJobId()));
        } finally {
            dao.delete(JOB5.getJobId());
        }
    }

    @Test
    public void testDelete() {
        dao.create(JOB1);
        assertTrue(dao.delete(JOB1.getJobId()));
        assertFalse(dao.delete(JobId.of("NoSuchJob")));
    }

    @Test
    public void testGetNextRunTime() {
        assertEquals(NEXT_RUN_DATE, dao.getNextRunTime(JobId.of("CleanupEmptyRescopesJob")));
        assertNull(dao.getNextRunTime(JobId.of("UserCleanupJob")));
        assertNull(dao.getNextRunTime(JobId.of("NoSuchJob")));
    }

    @Test
    public void testGetVersion() {
        assertEquals((Long) 1L, dao.getVersion(JobId.of("CleanupEmptyRescopesJob")));
        assertNull(dao.getVersion(JobId.of("NoSuchJob")));
    }

    @Test
    public void testFind() {
        assertNotNull(dao.find(JobId.of("CleanupEmptyRescopesJob")));
        assertNull(dao.getVersion(JobId.of("NoSuchJob")));
    }

    @Test
    public void testFindAllJobRunnerKeys() {
        Set<JobRunnerKey> jobRunnerKeys = dao.findAllJobRunnerKeys();
        assertEquals(2, jobRunnerKeys.size());
        assertTrue(jobRunnerKeys.contains(JobRunnerKey.of(
                "com.atlassian.stash.internal.pull.rescope.DefaultRescopeProcessor$CleanupEmptyRescopesJob")));
        assertTrue(jobRunnerKeys.contains(JobRunnerKey.of(
                "com.atlassian.stash.internal.user.DefaultUserAdminService$UserCleanupJob")));
    }

    @Test
    public void testFindByJobRunnerKey() {
        Collection<ClusteredJob> jobs = dao.findByJobRunnerKey(JobRunnerKey.of(
                "com.atlassian.stash.internal.pull.rescope.DefaultRescopeProcessor$CleanupEmptyRescopesJob"));
        assertEquals(1, jobs.size());
        assertEquals(JobId.of("CleanupEmptyRescopesJob"), jobs.iterator().next().getJobId());
    }

    @Test
    public void testRefresh() {
        Map<JobId, Date> jobMap = dao.refresh();
        assertEquals(1, jobMap.size());
        assertTrue(jobMap.keySet().contains(JobId.of("CleanupEmptyRescopesJob")));
        assertEquals(NEXT_RUN_DATE, jobMap.get(JobId.of("CleanupEmptyRescopesJob")));
    }

    @Test
    public void testUpdateNextRunTime() {
        dao.create(JOB1);
        try {
            Date nextRunTime = new Date(System.currentTimeMillis() + 60000L);
            assertFalse(dao.updateNextRunTime(JOB1.getJobId(), nextRunTime, 2L));   // version mismatch
            assertTrue(dao.updateNextRunTime(JOB1.getJobId(), nextRunTime, 0L));
            assertEquals(nextRunTime, dao.getNextRunTime(JOB1.getJobId()));
        } finally {
            dao.delete(JOB1.getJobId());
        }
    }

    /**
     * Microsoft SQL Server's "datetime" <em>rounds</em> timestamps to increments of .000, .003, or .007 seconds.  A
     * "datetime2" column type is supported in SQL Server 2008 but this is not currently used by default in Liquibase.
     * See https://msdn.microsoft.com/en-us/library/ms187819.aspx (especially "Rounding of datetime Fractional Second
     * Precision").
     *
     * MySQL prior to 5.6.4 also did not persist fractional seconds in timestamp values and <em>truncates</em> to the
     * next lowest whole second.  See http://dev.mysql.com/doc/refman/5.6/en/fractional-seconds.html.
     *
     * So to be liberal in what we accept, allow up to 1000 milliseconds of "slop" in all timestamp comparisons.
     */
    private static void assertEquals(Date expected, Date actual) {
        Assert.assertEquals(expected.getTime(), actual.getTime(), 1000L);
    }

}
