package com.atlassian.scheduler.quartz2;
   
   import com.atlassian.scheduler.SchedulerRuntimeException;
   import com.atlassian.scheduler.SchedulerService;
   import com.atlassian.scheduler.SchedulerServiceException;
   import com.atlassian.scheduler.config.JobConfig;
   import com.atlassian.scheduler.config.JobId;
   import com.atlassian.scheduler.config.JobRunnerKey;
   import com.atlassian.scheduler.config.RunMode;
  import com.atlassian.scheduler.config.Schedule;
  import com.atlassian.scheduler.core.AbstractSchedulerService;
  import com.atlassian.scheduler.core.JobLauncher;
  import com.atlassian.scheduler.core.spi.RunDetailsDao;
  import com.atlassian.scheduler.core.spi.SchedulerServiceConfiguration;
  import com.atlassian.scheduler.quartz2.spi.Quartz2SchedulerConfiguration;
  import com.atlassian.scheduler.status.JobDetails;
  import com.google.common.collect.ImmutableList;
  import com.google.common.collect.ImmutableSet;
  import org.quartz.Scheduler;
  import org.quartz.Trigger;
  import org.quartz.TriggerBuilder;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import javax.annotation.Nonnull;
  import javax.annotation.Nullable;
  import java.io.Serializable;
  import java.util.Comparator;
  import java.util.Date;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Set;
  
  import static com.atlassian.scheduler.config.RunMode.RUN_LOCALLY;
  import static com.atlassian.scheduler.config.RunMode.RUN_ONCE_PER_CLUSTER;
  import static com.atlassian.scheduler.quartz2.Quartz2SchedulerFacade.createClustered;
  import static com.atlassian.scheduler.quartz2.Quartz2SchedulerFacade.createLocal;
  import static com.atlassian.util.concurrent.Assertions.notNull;
  import static com.google.common.collect.Lists.newArrayList;
  import static com.google.common.collect.Sets.newHashSet;
  import static java.util.Collections.sort;
  import static org.quartz.TriggerKey.triggerKey;
  
  /**
   * Quartz 2.x implementation of a {@link SchedulerService}.
   * <ul>
   * <li>Job runner keys are mapped to Quartz {@code JobKey} names, with the {@code Job} being created or
   * destroyed automatically based on whether or not it has any existing {@code Trigger}s.</li>
   * <li>All Quartz {@code Job}s use {@link Quartz2Job}, which immediately delegates to {@link JobLauncher}.</li>
   * <li>Job ids are mapped to Quartz {@code TriggerKey} names.</li>
   * <li>The parameters map is serialized to a {@code byte[]} and stored in the {@code JobDataMap}
   * for the Quartz {@code Trigger}.</li>
   * </ul>
   */
  public class Quartz2SchedulerService extends AbstractSchedulerService {
      private static final Logger LOG = LoggerFactory.getLogger(Quartz2SchedulerService.class);
  
      private final Quartz2SchedulerFacade localJobs;
      private final Quartz2SchedulerFacade clusteredJobs;
      private final Quartz2TriggerFactory triggerFactory;
      private final Quartz2JobDetailsFactory jobDetailsFactory;
  
      @SuppressWarnings("unused")
      public Quartz2SchedulerService(final RunDetailsDao runDetailsDao, final Quartz2SchedulerConfiguration config)
              throws SchedulerServiceException {
          super(runDetailsDao);
          this.localJobs = createLocal(this, config);
          this.clusteredJobs = createClustered(this, config);
          this.triggerFactory = new Quartz2TriggerFactory(config, getParameterMapSerializer());
          this.jobDetailsFactory = new Quartz2JobDetailsFactory(this);
      }
  
      @SuppressWarnings("unused")
      public Quartz2SchedulerService(final RunDetailsDao runDetailsDao, final SchedulerServiceConfiguration config,
                                     final Scheduler localScheduler, final Scheduler clusteredScheduler)
              throws SchedulerServiceException {
          super(runDetailsDao);
          this.localJobs = createLocal(this, localScheduler);
          this.clusteredJobs = createClustered(this, clusteredScheduler);
          this.triggerFactory = new Quartz2TriggerFactory(config, getParameterMapSerializer());
          this.jobDetailsFactory = new Quartz2JobDetailsFactory(this);
      }
  
      @Override
      public void scheduleJob(final JobId jobId, final JobConfig jobConfig) throws SchedulerServiceException {
          try {
              notNull("jobConfig", jobConfig);
              LOG.debug("scheduleJob: {}: {}", jobId, jobConfig);
  
              localJobs.unscheduleJob(jobId);
              clusteredJobs.unscheduleJob(jobId);
  
              final Quartz2SchedulerFacade facade = getFacade(jobConfig.getRunMode());
              final JobRunnerKey jobRunnerKey = jobConfig.getJobRunnerKey();
              final TriggerBuilder<?> trigger = triggerFactory.buildTrigger(jobId, jobConfig);
              facade.scheduleJob(jobRunnerKey, trigger);
          } catch (SchedulerRuntimeException sre) {
              throw checked(sre);
          }
     }
 
 
     @Override
     public void unscheduleJob(final JobId jobId) {
         // Deliberately avoiding a short-circuit; we want to remove from both
         boolean found = localJobs.unscheduleJob(jobId);
         found |= clusteredJobs.unscheduleJob(jobId);
         if (found) {
             LOG.debug("unscheduleJob: {}", jobId);
         } else {
             LOG.debug("unscheduleJob for non-existent jobId: {}", jobId);
         }
     }
 
     @Nullable
     @Override
     public Date calculateNextRunTime(Schedule schedule) throws SchedulerServiceException {
         final Trigger trigger = triggerFactory.buildTrigger(schedule)
                 .withIdentity(triggerKey("name", "group"))
                 .build();
         return trigger.getFireTimeAfter(new Date());
     }
 
     @Nullable
     @Override
     public JobDetails getJobDetails(final JobId jobId) {
         JobDetails jobDetails = getJobDetails(clusteredJobs, jobId, RUN_ONCE_PER_CLUSTER);
         if (jobDetails == null) {
             jobDetails = getJobDetails(localJobs, jobId, RUN_LOCALLY);
         }
         return jobDetails;
     }
 
     @Nonnull
     @Override
     public Set<JobRunnerKey> getJobRunnerKeysForAllScheduledJobs() {
         final Set<JobRunnerKey> jobRunnerKeys = new HashSet<JobRunnerKey>(localJobs.getJobRunnerKeys());
         jobRunnerKeys.addAll(clusteredJobs.getJobRunnerKeys());
         return ImmutableSet.copyOf(jobRunnerKeys);
     }
 
     @Nullable
     private JobDetails getJobDetails(Quartz2SchedulerFacade facade, JobId jobId, RunMode runMode) {
         final Trigger trigger = facade.getTrigger(jobId);
         return (trigger != null) ? jobDetailsFactory.buildJobDetails(jobId, trigger, runMode) : null;
     }
 
     @Nonnull
     @Override
     public List<JobDetails> getJobsByJobRunnerKey(final JobRunnerKey jobRunnerKey) {
         final UniqueJobDetailsCollector collector = new UniqueJobDetailsCollector();
         collector.collect(RUN_ONCE_PER_CLUSTER, clusteredJobs.getTriggersOfJob(jobRunnerKey));
         collector.collect(RUN_LOCALLY, localJobs.getTriggersOfJob(jobRunnerKey));
         return collector.getResults();
     }
 
     /**
      * Start the threads associated with each quartz scheduler. It is the responsibility of the underlying
      * JobStores to determine whether or not to trigger jobs which should have run while the scheduler was
      * in standby mode. This is usually controlled by a misfire threshold on the JobStore implementation
      */
     @Override
     protected void startImpl() throws SchedulerServiceException {
         boolean abort = true;
         localJobs.start();
         try {
             clusteredJobs.start();
             abort = false;
         } finally {
             if (abort) {
                 localJobs.standby();
             }
         }
     }
 
     /**
      * Stop the threads associated with each quartz scheduler
      */
     @Override
     protected void standbyImpl() throws SchedulerServiceException {
         try {
             localJobs.standby();
         } finally {
             clusteredJobs.standby();
         }
     }
 
     @Override
     protected void shutdownImpl() {
         try {
             localJobs.shutdown();
         } finally {
             clusteredJobs.shutdown();
         }
     }
 
     private Quartz2SchedulerFacade getFacade(RunMode runMode) {
         switch (notNull("runMode", runMode)) {
             case RUN_LOCALLY:
                 return localJobs;
             case RUN_ONCE_PER_CLUSTER:
                 return clusteredJobs;
         }
         throw new IllegalArgumentException("runMode=" + runMode);
     }
 
 
     /**
      * Since it is possible for another node in the cluster to register the same jobId for running clustered
      * when we had it set to run locally, deal with this by pretending the local one is not there.
      */
     class UniqueJobDetailsCollector {
         final Set<String> jobIdsSeen = newHashSet();
         final List<JobDetails> jobs = newArrayList();
 
         void collect(RunMode runMode, final List<? extends Trigger> triggers) {
             for (Trigger trigger : triggers) {
                 final String jobId = trigger.getKey().getName();
                 if (jobIdsSeen.add(jobId)) {
                     try {
                         jobs.add(jobDetailsFactory.buildJobDetails(JobId.of(jobId), trigger, runMode));
                     } catch (SchedulerRuntimeException sre) {
                         LOG.debug("Unable to reconstruct log details for jobId '{}': {}", jobId, sre);
                     }
                 }
             }
         }
 
         List<JobDetails> getResults() {
             sort(jobs, new SortByJobId());
             return ImmutableList.copyOf(jobs);
         }
     }
 
     static class SortByJobId implements Comparator<JobDetails>, Serializable {
         private static final long serialVersionUID = 1L;
 
         @Override
         public int compare(final JobDetails jd1, final JobDetails jd2) {
             return jd1.getJobId().compareTo(jd2.getJobId());
         }
     }
 }