package com.atlassian.stash.internal.scheduling;

import com.atlassian.scheduler.caesium.spi.ClusteredJob;
import com.atlassian.scheduler.caesium.spi.ClusteredJobDao;
import com.atlassian.scheduler.config.JobId;
import com.atlassian.scheduler.config.JobRunnerKey;
import com.atlassian.stash.internal.spring.SpringTransactionUtils;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import org.hibernate.NonUniqueObjectException;
import org.hibernate.Session;
import org.hibernate.SessionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataIntegrityViolationException;
import org.springframework.stereotype.Repository;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.support.TransactionTemplate;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.sql.Timestamp;
import java.util.*;

/**
 * A Hibernate-based implementation of {@link ClusteredJobDao}.  Note this DAO must manage its own transactions as it
 * is intended to be called directly from Caesium, not from service layer classes which define their own transaction
 * boundaries.
 */
@Repository("clusteredJobDao")
public class HibernateClusteredJobDao implements ClusteredJobDao {

    private final SessionFactory sessionFactory;

    private static final Logger log = LoggerFactory.getLogger(HibernateClusteredJobDao.class);

    private final TransactionTemplate readOnlyTransaction;
    private final TransactionTemplate readWriteTransaction;

    @Autowired
    public HibernateClusteredJobDao(SessionFactory sessionFactory, PlatformTransactionManager transactionManager) {
        this.sessionFactory = sessionFactory;
        readOnlyTransaction = new TransactionTemplate(transactionManager, SpringTransactionUtils.REQUIRES_NEW);
        readOnlyTransaction.setReadOnly(true);
        readWriteTransaction = new TransactionTemplate(transactionManager, SpringTransactionUtils.REQUIRES_NEW);
    }

    @Override
    public boolean create(@Nonnull ClusteredJob clusteredJob) {
        try {
            return readWriteTransaction.execute(action -> {
                session().save(new InternalClusteredJob.Builder(clusteredJob).build());
                return true;
            });
        } catch (DataIntegrityViolationException | NonUniqueObjectException e) {
            return false;
        }
    }

    @Override
    public boolean delete(@Nonnull JobId jobId) {
        int rows = readWriteTransaction.execute(action ->
            session().createQuery("delete from InternalClusteredJob where jobId = :jobId")
                    .setParameter("jobId", jobId.toString())
                    .executeUpdate());
        return rows > 0;
    }

    @Nullable
    @Override
    public ClusteredJob find(@Nonnull JobId jobId) {
        return readOnlyTransaction.execute(status ->
                (InternalClusteredJob) session().get(InternalClusteredJob.class, jobId.toString()));
    }

    @Nonnull
    @Override
    public Set<JobRunnerKey> findAllJobRunnerKeys() {
        List<?> results = readOnlyTransaction.execute(status ->
                session().createQuery("select distinct jobRunnerKey from InternalClusteredJob")
                        .list());
        ImmutableSet.Builder<JobRunnerKey> keys = ImmutableSet.builder();
        for (Object result : results) {
            keys.add(JobRunnerKey.of((String) result));
        }
        return keys.build();
    }

    @Nonnull
    @Override
    public Collection<ClusteredJob> findByJobRunnerKey(@Nonnull JobRunnerKey jobRunnerKey) {
        List<?> results = readOnlyTransaction.execute(status ->
                session().createQuery("from InternalClusteredJob where jobRunnerKey = :jobRunnerKey")
                        .setParameter("jobRunnerKey", jobRunnerKey.toString())
                        .list());
        ImmutableSet.Builder<ClusteredJob> clusteredJobs = ImmutableSet.builder();
        for (Object result : results) {
            clusteredJobs.add((InternalClusteredJob) result);
        }
        return clusteredJobs.build();
    }

    @Nullable
    @Override
    public Date getNextRunTime(@Nonnull JobId jobId) {
        Object result = readOnlyTransaction.execute(status ->
                session().createQuery("select nextRun from InternalClusteredJob where jobId = :jobId")
                        .setParameter("jobId", jobId.toString())
                        .uniqueResult());
        return result == null ? null : new Date(((Timestamp) result).getTime());
    }

    @Nullable
    @Override
    public Long getVersion(@Nonnull JobId jobId) {
        return (Long) readOnlyTransaction.execute(status ->
                session().createQuery("select version from InternalClusteredJob where jobId = :jobId")
                        .setParameter("jobId", jobId.toString())
                        .uniqueResult());
    }

    @Nonnull
    @Override
    public Map<JobId, Date> refresh() {
        List<?> results = readOnlyTransaction.execute(status ->
                session().createQuery("select jobId, nextRun from InternalClusteredJob where nextRun is not null")
                        .list());
        final ImmutableMap.Builder<JobId, Date> jobMap = ImmutableMap.builder();
        for (Object result : results) {
            Object[] columns = (Object[]) result;
            jobMap.put(JobId.of((String) columns[0]), (Date) columns[1]);
        }
        return jobMap.build();
    }

    @Override
    public boolean updateNextRunTime(@Nonnull JobId jobId, @Nullable final Date date, long version) {
        int rows = readWriteTransaction.execute(status ->
                session().createQuery("update InternalClusteredJob set nextRun = :nextRun, version = :newVersion " +
                        "where jobId = :jobId and version = :version")
                        .setParameter("nextRun", date)
                        .setParameter("newVersion", version + 1L)
                        .setParameter("jobId", jobId.toString())
                        .setParameter("version", version)
                        .executeUpdate());
        return rows > 0;
    }

    private Session session() {
        return sessionFactory.getCurrentSession();
    }

}