/*
 * Copyright 2017 LinkedIn Corp.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */

package azkaban.scheduler;

import static azkaban.ServiceProvider.SERVICE_PROVIDER;
import static java.util.Objects.requireNonNull;

import azkaban.Constants.ConfigurationKeys;
import azkaban.utils.Props;
import java.util.List;
import javax.inject.Inject;
import javax.inject.Singleton;
import org.quartz.CronExpression;
import org.quartz.CronScheduleBuilder;
import org.quartz.JobBuilder;
import org.quartz.JobDetail;
import org.quartz.JobKey;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.quartz.Trigger;
import org.quartz.Trigger.TriggerState;
import org.quartz.TriggerBuilder;
import org.quartz.impl.StdSchedulerFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Manages Quartz schedules. Azkaban regards QuartzJob and QuartzTrigger as an one-to-one
 * mapping.
 * Quartz job key naming standard:
 * Job key is composed of job name and group name. Job type denotes job name. Project id+flow
 * name denotes group name.
 * E.x FLOW_TRIGGER as job name, 1.flow1 as group name
 */
@Singleton
public class QuartzScheduler {

  private static final Logger logger = LoggerFactory.getLogger(QuartzScheduler.class);
  private Scheduler scheduler = null;

  @Inject
  public QuartzScheduler(final Props azProps) throws SchedulerException {
    if (!azProps.getBoolean(ConfigurationKeys.ENABLE_QUARTZ, false)) {
      return;
    }
    // TODO kunkun-tang: Many quartz properties should be defaulted such that not necessarily being
    // checked into azkaban.properties. Also, we need to only assemble Quartz related properties
    // here, which should be done in Azkaban WebServer Guice Module.
    final StdSchedulerFactory schedulerFactory =
        new StdSchedulerFactory(azProps.toAllProperties());
    this.scheduler = schedulerFactory.getScheduler();

    // Currently Quartz only support internal job schedules. When we migrate to User Production
    // flows, we need to construct a Guice-Free JobFactory for use.
    this.scheduler.setJobFactory(SERVICE_PROVIDER.getInstance(SchedulerJobFactory.class));
  }

  public void start() {
    try {
      this.scheduler.start();
    } catch (final SchedulerException e) {
      logger.error("Error starting Quartz scheduler: ", e);
    }
    logger.info("Quartz Scheduler started.");
  }

  public void cleanup() {
    logger.info("Cleaning up schedules in scheduler");
    try {
      this.scheduler.clear();
    } catch (final SchedulerException e) {
      logger.error("Exception clearing scheduler: ", e);
    }
  }

  public void pause() {
    logger.info("pausing all schedules in Quartz");
    try {
      this.scheduler.pauseAll();
    } catch (final SchedulerException e) {
      logger.error("Exception pausing scheduler: ", e);
    }
  }

  public void resume() {
    logger.info("resuming all schedules in Quartz");
    try {
      this.scheduler.resumeAll();
    } catch (final SchedulerException e) {
      logger.error("Exception resuming scheduler: ", e);
    }
  }

  public void shutdown() {
    logger.info("Shutting down scheduler");
    try {
      this.scheduler.shutdown();
    } catch (final SchedulerException e) {
      logger.error("Exception shutting down scheduler: ", e);
    }
  }

  private void checkJobExistence(final String jobName, final String groupName)
      throws SchedulerException {
    if (!ifJobExist(jobName, groupName)) {
      throw new SchedulerException(String.format("can not find job with job name: %s and group "
          + "name %s: in quartz.", jobName, groupName));
    }
  }


  /**
   * pause a job given the groupname. since pausing request might be issued concurrently,
   * so synchronized is added to ensure thread safety.
   */
  public synchronized void pauseJob(final String jobName, final String groupName)
      throws SchedulerException {
    checkJobExistence(jobName, groupName);
    this.scheduler.pauseJob(new JobKey(jobName, groupName));
  }

  public synchronized boolean isJobPaused(final String jobName, final String groupName)
      throws SchedulerException {
    final JobKey jobKey = new JobKey(jobName, groupName);
    final JobDetail jobDetail = this.scheduler.getJobDetail(jobKey);
    final List<? extends Trigger> triggers = this.scheduler.getTriggersOfJob(jobDetail.getKey());
    for (final Trigger trigger : triggers) {
      final TriggerState triggerState = this.scheduler.getTriggerState(trigger.getKey());
      if (TriggerState.PAUSED.equals(triggerState)) {
        return true;
      }
    }
    return false;
  }

  /**
   * resume a paused job given the groupname. since resuming request might be issued concurrently,
   * so synchronized is added to ensure thread safety.
   */
  public synchronized void resumeJob(final String jobName, final String groupName)
      throws SchedulerException {
    checkJobExistence(jobName, groupName);
    this.scheduler.resumeJob(new JobKey(jobName, groupName));
  }

  /**
   * Unregister a job given the groupname. Since unregister might be called when
   * concurrently removing projects, so synchronized is added to ensure thread safety.
   */
  public synchronized void unregisterJob(final String jobName, final String groupName) throws
      SchedulerException {
    checkJobExistence(jobName, groupName);
    this.scheduler.deleteJob(new JobKey(jobName, groupName));
  }

  /**
   * Only cron schedule register is supported. Since register might be called when
   * concurrently uploading projects, so synchronized is added to ensure thread safety.
   *
   * @param cronExpression the cron schedule for this job
   * @param jobDescription Regarding QuartzJobDescription#groupName, in order to guarantee no
   * duplicate quartz schedules, we design the naming convention depending on use cases: <ul>
   * <li>User flow schedule: we use {@link JobKey#JobKey} to represent the identity of a
   * flow's schedule. The format follows "$projectID.$flowName" to guarantee no duplicates.
   * <li>Quartz schedule for AZ internal use: the groupName should start with letters, rather
   * than
   * number, which is the first case.</ul>
   */
  public synchronized void registerJob(final String cronExpression, final QuartzJobDescription
      jobDescription)
      throws SchedulerException {

    requireNonNull(jobDescription, "jobDescription is null");

    if (ifJobExist(jobDescription.getJobName(), jobDescription.getGroupName())) {
      throw new SchedulerException(String.format("can not register existing job with job name: "
          + "%s and group name: %s", jobDescription.getJobName(), jobDescription.getGroupName()));
    }

    if (!CronExpression.isValidExpression(cronExpression)) {
      throw new SchedulerException(
          "The cron expression string <" + cronExpression + "> is not valid.");
    }

    // TODO kunkun-tang: we will modify this when we start supporting multi schedules per flow.
    final JobDetail job = JobBuilder.newJob(jobDescription.getJobClass())
        .withIdentity(jobDescription.getJobName(), jobDescription.getGroupName()).build();

    // Add external dependencies to Job Data Map.
    job.getJobDataMap().putAll(jobDescription.getContextMap());

    // TODO kunkun-tang: Need management code to deal with different misfire policy
    final Trigger trigger = TriggerBuilder
        .newTrigger()
        .withSchedule(
            CronScheduleBuilder.cronSchedule(cronExpression)
                .withMisfireHandlingInstructionFireAndProceed()
//            .withMisfireHandlingInstructionDoNothing()
//            .withMisfireHandlingInstructionIgnoreMisfires()
        )
        .build();

    this.scheduler.scheduleJob(job, trigger);
    logger.info("Quartz Schedule with jobDetail " + job.getDescription() + " is registered.");
  }


  public boolean ifJobExist(final String jobName, final String groupName)
      throws SchedulerException {
    return this.scheduler.getJobDetail(new JobKey(jobName, groupName)) != null;
  }

  public Scheduler getScheduler() {
    return this.scheduler;
  }
}
