package com.redislabs.quartz;

import com.github.jedis.lock.JedisLock;
import com.google.gson.Gson;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.Reader;
import java.io.Writer;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import org.quartz.Calendar;
import org.quartz.CronTrigger;
import org.quartz.DisallowConcurrentExecution;
import org.quartz.Job;
import org.quartz.JobBuilder;
import org.quartz.JobDataMap;
import org.quartz.JobDetail;
import org.quartz.JobKey;
import org.quartz.JobPersistenceException;
import org.quartz.ObjectAlreadyExistsException;
import org.quartz.PersistJobDataAfterExecution;
import org.quartz.SchedulerConfigException;
import org.quartz.SchedulerException;
import org.quartz.SimpleTrigger;
import org.quartz.Trigger;
import org.quartz.Trigger.CompletedExecutionInstruction;
import org.quartz.Trigger.TriggerState;
import org.quartz.TriggerKey;
import org.quartz.impl.matchers.GroupMatcher;
import org.quartz.impl.matchers.StringMatcher.StringOperatorName;
import org.quartz.impl.triggers.CronTriggerImpl;
import org.quartz.impl.triggers.SimpleTriggerImpl;
import org.quartz.spi.ClassLoadHelper;
import org.quartz.spi.JobStore;
import org.quartz.spi.OperableTrigger;
import org.quartz.spi.SchedulerSignaler;
import org.quartz.spi.TriggerFiredBundle;
import org.quartz.spi.TriggerFiredResult;
import org.quartz.utils.ClassUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.JedisPubSub;
import redis.clients.jedis.Pipeline;
import redis.clients.jedis.Protocol;
import redis.clients.jedis.Tuple;

/**
 * <p>
 * This class implements a <code>{@link org.quartz.spi.JobStore}</code> that
 * utilizes <a href="http://redis.io/">Redis</a> as a persistent storage.
 * </p>
 * 
 * @author Matan Kehat
 */
public class RedisJobStore implements JobStore {
	
	/*** Sets Keys ***/
	private static final String JOBS_SET = "jobs";
	private static final String JOB_GROUPS_SET = "job_groups";
	private static final String BLOCKED_JOBS_SET = "blocked_jobs";
	private static final String TRIGGERS_SET = "triggers";
	private static final String TRIGGER_GROUPS_SET = "trigger_groups";
	private static final String CALENDARS_SET = "calendars";
	private static final String PAUSED_TRIGGER_GROUPS_SET = "paused__trigger_groups";
	private static final String PAUSED_JOB_GROUPS_SET = "paused_job_groups";
		
	/*** Job Hash Fields ***/
	private static final String JOB_CLASS = "job_class_name";
	private static final String DESCRIPTION = "description";
	private static final String IS_DURABLE = "is_durable";
	private static final String BLOCKED_BY = "blocked_by";
	private static final String BLOCK_TIME = "block_time";
		
	/*** Trigger Hash Fields ***/
	private static final String JOB_HASH_KEY = "job_hash_key";
	private static final String NEXT_FIRE_TIME = "next_fire_time";
	private static final String PREV_FIRE_TIME = "prev_fire_time";
	private static final String PRIORITY = "priority";
	private static final String TRIGGER_TYPE = "trigger_type";
	private static final String CALENDAR_NAME = "calendar_name";
	private static final String START_TIME = "start_time";
	private static final String END_TIME = "end_time";
	private static final String FINAL_FIRE_TIME = "final_fire_time";
	private static final String FIRE_INSTANCE_ID = "fire_instance_id";
	private static final String MISFIRE_INSTRUCTION = "misfire_instruction";
	private static final String LOCKED_BY = "locked_by";
	private static final String LOCK_TIME = "lock_time";
		
	/*** Simple Trigger Fields ***/
	private static final String TRIGGER_TYPE_SIMPLE = "SIMPLE";
	private static final String REPEAT_COUNT = "repeat_count";
	private static final String REPEAT_INTERVAL = "repeat_interval";
	private static final String TIMES_TRIGGERED = "times_triggered";
	
	/*** Cron Trigger Fields ***/
	private static final String TRIGGER_TYPE_CRON = "CRON";
	private static final String CRON_EXPRESSION = "cron_expression";
	private static final String TIME_ZONE_ID = "time_zone_id";
	
	/*** Calendar Hash Fields ***/	
	private static final String CALENDAR_CLASS = "calendar_class";
	private static final String CALENDAR_SERIALIZED = "calendar_serialized";
	
	/*** Misc Keys ***/
	private static final String LAST_TRIGGERS_RELEASE_TIME = "last_triggers_release_time";
	private static final String LOCK = "lock";
	
	/*** Channels ***/
	private static final String UNLOCK_NOTIFICATIONS_CHANNEL = "unlock-notificactions";
	 
	/*** Class Members ***/
	private final Logger log = LoggerFactory.getLogger(getClass());
	private static JedisPool pool;	
   private static JedisLock lockPool;
	private ClassLoadHelper loadHelper;
	private SchedulerSignaler signaler;
	private String instanceId;
	private String host;
	private int port;
	private String password;
	private String instanceIdFilePath;
	private int releaseTriggersInterval; // In seconds 
	private int lockTimeout; 
	
	protected long misfireThreshold = 60000L;
		
	
	@Override
	public void initialize(ClassLoadHelper loadHelper,
			SchedulerSignaler signaler) throws SchedulerConfigException {
		
		this.loadHelper = loadHelper;
		this.signaler = signaler;		
	   
    	// initializing a connection pool
    	JedisPoolConfig config = new JedisPoolConfig();
    	if (password != null)
    		pool = new JedisPool(config, host, port, Protocol.DEFAULT_TIMEOUT, password);	    		
    	else
    		pool = new JedisPool(config, host, port, Protocol.DEFAULT_TIMEOUT);
    	
    	// initializing a locking connection pool with a longer timeout
    	if (lockTimeout == 0)
    		lockTimeout = 10 * 60 * 1000; // 10 Minutes locking timeout
    	
      lockPool = new JedisLock(pool.getResource(), "JobLock", lockTimeout);

	}	

	@Override
	public void schedulerStarted() throws SchedulerException {
		if (this.instanceIdFilePath != null) {
			String path = this.instanceIdFilePath + File.separator + "scheduler.id";
			File instanceIdFile = new File(path);
			
			// releasing triggers that a previous scheduler instance didn't release		
			String lockedByInstanceId = readInstanceId(instanceIdFile);
			if (lockedByInstanceId != null) {
				releaseLockedTriggers(lockedByInstanceId, RedisTriggerState.ACQUIRED, RedisTriggerState.WAITING);
				releaseLockedTriggers(lockedByInstanceId, RedisTriggerState.BLOCKED, RedisTriggerState.WAITING);
				releaseLockedTriggers(lockedByInstanceId, RedisTriggerState.PAUSED_BLOCKED, RedisTriggerState.PAUSED);
				
				releaseBlockedJobs(lockedByInstanceId);
			}
			
			// writing the current instance id to a file
			writeInstanceId(instanceIdFile);			
		}			
	}

	@Override
	public void schedulerPaused() {
		// No-op
	}

	@Override
	public void schedulerResumed() {
		// No-op
	}

	@Override
	public void shutdown() {
		if (pool != null)
			pool.destroy();
   }

	@Override
	public boolean supportsPersistence() {
		return true;
	}

	@Override
	public long getEstimatedTimeToReleaseAndAcquireTrigger() {
		// varied
		return 100;
	}

	@Override
	public boolean isClustered() {
		return true;
	}

	@Override
	public void storeJobAndTrigger(JobDetail newJob, OperableTrigger newTrigger)
			throws ObjectAlreadyExistsException, JobPersistenceException {
		String jobHashKey = createJobHashKey(newJob.getKey().getGroup(), newJob.getKey().getName());
		String triggerHashKey = createTriggerHashKey(newTrigger.getKey().getGroup(), newTrigger.getKey().getName());
      try (Jedis jedis = pool.getResource()) {
         lockPool.acquire();
			storeJob(newJob, false, jedis);
			storeTrigger(newTrigger, false, jedis);			
      } catch (ObjectAlreadyExistsException ex) {
         log.warn("could not store job: " + jobHashKey + " and trigger: " + triggerHashKey, ex);
      } catch (Exception ex) {
         log.error("could not store job: " + jobHashKey + " and trigger: " + triggerHashKey, ex);
         throw new JobPersistenceException(ex.getMessage(), ex.getCause());
      } finally {
         lockPool.release();
		}		
	}

	@Override
   public void storeJob(JobDetail newJob, boolean replaceExisting) throws JobPersistenceException {
		String jobHashKey = createJobHashKey(newJob.getKey().getGroup(), newJob.getKey().getName());
      try (Jedis jedis = pool.getResource()) {
         lockPool.acquire();
			storeJob(newJob, replaceExisting, jedis);
		} catch (ObjectAlreadyExistsException ex) {
			log.warn(jobHashKey + " already exists", ex);
			throw ex;
		} catch (Exception ex) {
			log.error("could not store job: " + jobHashKey, ex);
			throw new JobPersistenceException(ex.getMessage(), ex.getCause());
		} finally {
         lockPool.release();
		}		
	}

	/**
	 * Stores job in redis.
	 *
	 * @param newJob the new job
	 * @param replaceExisting the replace existing
	 * @param jedis thread-safe redis connection
	 * @throws ObjectAlreadyExistsException
	 */
	private void storeJob(JobDetail newJob, boolean replaceExisting, Jedis jedis)
			throws ObjectAlreadyExistsException {
		String jobHashKey = createJobHashKey(newJob.getKey().getGroup(), newJob.getKey().getName());
		String jobDataMapHashKey = createJobDataMapHashKey(newJob.getKey().getGroup(), newJob.getKey().getName());
		String jobGroupSetKey = createJobGroupSetKey(newJob.getKey().getGroup());
		
		if (jedis.exists(jobHashKey) && !replaceExisting)
			throw new ObjectAlreadyExistsException(newJob);
					
      Map<String, String> jobDetails = new HashMap<>();
		jobDetails.put(DESCRIPTION, newJob.getDescription() != null ? newJob.getDescription() : "");
		jobDetails.put(JOB_CLASS, newJob.getJobClass().getName());
		jobDetails.put(IS_DURABLE, Boolean.toString(newJob.isDurable()));
		jobDetails.put(BLOCKED_BY, "");
		jobDetails.put(BLOCK_TIME, "");
		jedis.hmset(jobHashKey, jobDetails);
		
		if (newJob.getJobDataMap() != null && !newJob.getJobDataMap().isEmpty())
			jedis.hmset(jobDataMapHashKey, getStringDataMap(newJob.getJobDataMap()));			
		
		jedis.sadd(JOBS_SET, jobHashKey);
		jedis.sadd(JOB_GROUPS_SET, jobGroupSetKey);
		jedis.sadd(jobGroupSetKey, jobHashKey);
	}
	
	/**
	 * Creates a job data map key in the form of: 'job_data_map:&#60;job_group_name&#62;:&#60;job_name&#62;'
	 *
	 * @param groupName the job group name
	 * @param jobName the job name
	 * @return the job data map key
	 */
	private String createJobDataMapHashKey(String groupName, String jobName) {
		return "job_data_map:" + groupName + ":" + jobName;
	}

	/**
	 * Creates a job group set key in the form of: 'job_group:&#60;job_group_name&#62;'
	 *
	 * @param groupName the group name
	 * @return the job group set key
	 */
	private String createJobGroupSetKey(String groupName) {
		return "job_group:" + groupName;
	}

	/**
	 * Creates a job key in the form of: 'job:&#60;job_group_name&#62;:&#60;job_name&#62;'
	 *
	 * @param groupName the job group name
	 * @param jobName the job name
	 * @return the job key
	 */
	private String createJobHashKey(String groupName, String jobName) {
		return "job:" + groupName + ":" + jobName;
	}
	
	/**
	 * Creates a calendar hash key in the form of: 'calendar:&#60;calendar_name&#62;'
	 *
	 * @param calendarName the calendar name
	 * @return the calendar key
	 */
	private String createCalendarHashKey(String calendarName) {
		return "calendar:" + calendarName;
	}

	private Map<String, String> getStringDataMap(JobDataMap jobDataMap) {
      Map<String, String> stringDataMap = new HashMap<>();
		for (String key : jobDataMap.keySet())
			stringDataMap.put(key, jobDataMap.get(key).toString());
					
		return stringDataMap;
	}

	@Override
	public void storeJobsAndTriggers(
			Map<JobDetail, Set<? extends Trigger>> triggersAndJobs,
			boolean replace) throws ObjectAlreadyExistsException,
			JobPersistenceException {
      try (Jedis jedis = pool.getResource()) {
         lockPool.acquire();
			
			// verifying jobs and triggers don't exist
			if (!replace) {
				for (Entry<JobDetail, Set<? extends Trigger>> entry : triggersAndJobs.entrySet()) {
					String jobHashKey = createJobHashKey(entry.getKey().getKey().getGroup(), entry.getKey().getKey().getName());
					if (jedis.exists(jobHashKey))
						throw new ObjectAlreadyExistsException(entry.getKey());
					
					for (Trigger trigger : entry.getValue()) {
						String triggerHashKey = createTriggerHashKey(trigger.getKey().getGroup(), trigger.getKey().getName());
						if (jedis.exists(triggerHashKey))
							throw new ObjectAlreadyExistsException(trigger);
					}
				}
			}
			
			// adding jobs and triggers
			for (Entry<JobDetail, Set<? extends Trigger>> entry : triggersAndJobs.entrySet()) {
				storeJob(entry.getKey(), true, jedis);
				for (Trigger trigger: entry.getValue())
					storeTrigger((OperableTrigger) trigger, true, jedis);
			}			
		} catch (Exception ex) {
			log.error("could not store jobs and triggers", ex);
			throw new JobPersistenceException(ex.getMessage(), ex.getCause());
		} finally {
         lockPool.release();
		}
	}

	@Override
	public boolean removeJob(JobKey jobKey) throws JobPersistenceException {
		boolean removed = false;
		String jobHashKey = createJobHashKey(jobKey.getGroup(), jobKey.getName());
      try (Jedis jedis = pool.getResource()) {
         lockPool.acquire();
			if (jedis.exists(jobHashKey)) {
				removeJob(jobKey, jedis);
				removed = true;
			}			
		} catch (Exception ex) {
			log.error("could not remove job: " + jobHashKey, ex);
			throw new JobPersistenceException(ex.getMessage(), ex.getCause());
		} finally {
         lockPool.release();
		}
		return removed;
	}

	/**
	 * Removes the job from redis.
	 *
	 * @param jobKey the job key
	 * @param jedis thread-safe redis connection
	 */
	private void removeJob(JobKey jobKey, Jedis jedis) {
		String jobHashKey = createJobHashKey(jobKey.getGroup(), jobKey.getName());
		String jobDataMapHashKey = createJobDataMapHashKey(jobKey.getGroup(), jobKey.getName());
		String jobGroupSetKey = createJobGroupSetKey(jobKey.getGroup());
		
		jedis.del(jobHashKey);
		jedis.del(jobDataMapHashKey);
		jedis.srem(JOBS_SET, jobHashKey);
		jedis.srem(BLOCKED_JOBS_SET, jobHashKey);
		jedis.srem(jobGroupSetKey, jobHashKey);
		if (jedis.scard(jobGroupSetKey) == 0) {
			jedis.srem(JOB_GROUPS_SET, jobGroupSetKey);
		}
	}

	@Override
	public boolean removeJobs(List<JobKey> jobKeys)
			throws JobPersistenceException {
      boolean removed = jobKeys.size() > 0;
		for (JobKey jobKey : jobKeys)
			removed = removed && removeJob(jobKey);
		
		return removed;
	}
	
	@Override
	public JobDetail retrieveJob(JobKey jobKey) throws JobPersistenceException {
		JobDetail jobDetail = null;
		String jobHashkey = createJobHashKey(jobKey.getGroup(), jobKey.getName());
      try (Jedis jedis = pool.getResource()) {
         lockPool.acquire();
			jobDetail = retrieveJob(jobKey, jedis);			
      } catch (JobPersistenceException | ClassNotFoundException | InterruptedException ex) {
			log.error("could not retrieve job: " + jobHashkey, ex);
			throw new JobPersistenceException(ex.getMessage(), ex.getCause());
		} finally {
         lockPool.release();
		}
		return jobDetail;
	}

	/**
	 * Retrieves job from redis.
	 *
	 * @param jobKey the job key
	 * @param jedis thread-safe redis connection
	 * @return the job detail
	 * @throws JobPersistenceException
	 */
	@SuppressWarnings("unchecked")
	private JobDetail retrieveJob(JobKey jobKey, Jedis jedis) throws JobPersistenceException, ClassNotFoundException {
		String jobHashkey = createJobHashKey(jobKey.getGroup(), jobKey.getName());
		String jobDataMapHashKey = createJobDataMapHashKey(jobKey.getGroup(), jobKey.getName());
      if (!jedis.exists(jobHashkey)) {
         log.warn("job: " + jobHashkey + " does not exist");
         return null;
      }
		Class<Job> jobClass = (Class<Job>) loadHelper.getClassLoader().loadClass(jedis.hget(jobHashkey, JOB_CLASS));
		JobBuilder jobBuilder = JobBuilder.newJob(jobClass)
									.withIdentity(jobKey)
									.withDescription(jedis.hget(jobHashkey, DESCRIPTION))
									.storeDurably(Boolean.getBoolean(jedis.hget(jobHashkey, IS_DURABLE)));
								
		Set<String> jobDataMapFields = jedis.hkeys(jobDataMapHashKey);
		if (!jobDataMapFields.isEmpty()) {
			for (String jobDataMapField : jobDataMapFields)
				jobBuilder.usingJobData(jobDataMapField, jedis.hget(jobDataMapHashKey, jobDataMapField));							
		}
		
		return jobBuilder.build();		
	}

	@Override
	public void storeTrigger(OperableTrigger newTrigger, boolean replaceExisting)
			throws ObjectAlreadyExistsException, JobPersistenceException {
		String triggerHashKey = createTriggerHashKey(newTrigger.getKey().getGroup(), newTrigger.getKey().getName());
      try (Jedis jedis = pool.getResource()) {
         lockPool.acquire();
			storeTrigger(newTrigger, replaceExisting, jedis);
		} catch (ObjectAlreadyExistsException ex) {
			log.warn(triggerHashKey + " already exists", ex);
			throw ex;
		} catch (Exception ex) {
			log.error("could not store trigger: " + triggerHashKey, ex);
			throw new JobPersistenceException(ex.getMessage(), ex.getCause());
		} finally {
         lockPool.release();
		}
	}

	/**
	 * Stores trigger in redis.
	 *
	 * @param newTrigger the new trigger
	 * @param replaceExisting replace existing
	 * @param jedis thread-safe redis connection
	 * @throws JobPersistenceException
	 * @throws ObjectAlreadyExistsException
	 */
	private void storeTrigger(OperableTrigger newTrigger, boolean replaceExisting, Jedis jedis)
           throws JobPersistenceException {
		String triggerHashKey = createTriggerHashKey(newTrigger.getKey().getGroup(), newTrigger.getKey().getName());
		String triggerGroupSetKey = createTriggerGroupSetKey(newTrigger.getKey().getGroup());
		String jobHashkey = createJobHashKey(newTrigger.getJobKey().getGroup(), newTrigger.getJobKey().getName());
		String jobTriggerSetkey = createJobTriggersSetKey(newTrigger.getJobKey().getGroup(), newTrigger.getJobKey().getName());
		
      if (jedis.exists(triggerHashKey) && !replaceExisting) {
         ObjectAlreadyExistsException ex = new ObjectAlreadyExistsException(newTrigger);
         log.warn(ex.toString());
      }
      Map<String, String> trigger = new HashMap<>();
		trigger.put(JOB_HASH_KEY, jobHashkey);
		trigger.put(DESCRIPTION, newTrigger.getDescription() != null ? newTrigger.getDescription() : "");
		trigger.put(NEXT_FIRE_TIME, newTrigger.getNextFireTime() != null ? Long.toString(newTrigger.getNextFireTime().getTime()) : "");
		trigger.put(PREV_FIRE_TIME, newTrigger.getPreviousFireTime() != null ? Long.toString(newTrigger.getPreviousFireTime().getTime()) : "");
		trigger.put(PRIORITY, Integer.toString(newTrigger.getPriority()));
		trigger.put(START_TIME, newTrigger.getStartTime() != null ? Long.toString(newTrigger.getStartTime().getTime()) : "");
		trigger.put(END_TIME, newTrigger.getEndTime() != null ? Long.toString(newTrigger.getEndTime().getTime()) : "");
		trigger.put(FINAL_FIRE_TIME, newTrigger.getFinalFireTime() != null ? Long.toString(newTrigger.getFinalFireTime().getTime()) : "");
		trigger.put(FIRE_INSTANCE_ID, newTrigger.getFireInstanceId() != null ?  newTrigger.getFireInstanceId() : "");
		trigger.put(MISFIRE_INSTRUCTION, Integer.toString(newTrigger.getMisfireInstruction()));
		trigger.put(CALENDAR_NAME, newTrigger.getCalendarName() != null ? newTrigger.getCalendarName() : "");
		if (newTrigger instanceof SimpleTrigger) {
			trigger.put(TRIGGER_TYPE, TRIGGER_TYPE_SIMPLE);
			trigger.put(REPEAT_COUNT, Integer.toString(((SimpleTrigger) newTrigger).getRepeatCount()));
			trigger.put(REPEAT_INTERVAL, Long.toString(((SimpleTrigger) newTrigger).getRepeatInterval()));
			trigger.put(TIMES_TRIGGERED, Integer.toString(((SimpleTrigger) newTrigger).getTimesTriggered()));
		} else if (newTrigger instanceof CronTrigger) {
			trigger.put(TRIGGER_TYPE, TRIGGER_TYPE_CRON);
			trigger.put(CRON_EXPRESSION, ((CronTrigger) newTrigger).getCronExpression() != null ? ((CronTrigger) newTrigger).getCronExpression() : "");
			trigger.put(TIME_ZONE_ID, ((CronTrigger) newTrigger).getTimeZone().getID() != null ? ((CronTrigger) newTrigger).getTimeZone().getID() : "");
		} else { // other trigger types are not supported
			 throw new UnsupportedOperationException();
		}		
		
		jedis.hmset(triggerHashKey, trigger);
		jedis.sadd(TRIGGERS_SET, triggerHashKey);
		jedis.sadd(TRIGGER_GROUPS_SET, triggerGroupSetKey);			
		jedis.sadd(triggerGroupSetKey, triggerHashKey);
		jedis.sadd(jobTriggerSetkey, triggerHashKey);
		if (newTrigger.getCalendarName() != null && !newTrigger.getCalendarName().isEmpty()) { // storing the trigger for calendar, if exists
			String calendarTriggersSetKey = createCalendarTriggersSetKey(newTrigger.getCalendarName());
			jedis.sadd(calendarTriggersSetKey, triggerHashKey);
		}
		
		if (jedis.sismember(PAUSED_TRIGGER_GROUPS_SET, triggerGroupSetKey) || jedis.sismember(PAUSED_JOB_GROUPS_SET, createJobGroupSetKey(newTrigger.getJobKey().getGroup()))) {	
			long nextFireTime = newTrigger.getNextFireTime() != null ? newTrigger.getNextFireTime().getTime() : -1;
			if (jedis.sismember(BLOCKED_JOBS_SET, jobHashkey))
				setTriggerState(RedisTriggerState.PAUSED_BLOCKED, (double)nextFireTime, triggerHashKey);
			else
				setTriggerState(RedisTriggerState.PAUSED, (double)nextFireTime, triggerHashKey);
		} else if (newTrigger.getNextFireTime() != null) {			
			setTriggerState(RedisTriggerState.WAITING, (double)newTrigger.getNextFireTime().getTime(), triggerHashKey);
		}
	}

	/**
	 * Creates a calendar_triggers hash key in the form of: <br>
	 * 'calendar_triggers:&#60;calendar_name&#62;'
	 *
	 * @param calendarName the calendar name
	 * @return the calendar_triggers set key
	 */
	private String createCalendarTriggersSetKey(String calendarName) {
		return "calendar_triggers:" + calendarName;
	}

	/**
	 * Creates a job_triggers set key in the form of: <br>
	 * 'job_triggers:&#60;job_group_name&#62;:&#60;job_name&#62;'
	 *
	 * @param jobGroupName the job group name
	 * @param jobName the job name
	 * @return the job_triggers set key
	 */
	private String createJobTriggersSetKey(String jobGroupName, String jobName) {
		return "job_triggers:" + jobGroupName + ":" + jobName;
	}

	/**
	 * Creates a trigger group hash key in the form of: <br>
	 * 'trigger_group:&#60;trigger_group_name&#62;'
	 *
	 * @param groupName the group name
	 * @return the trigger group set key
	 */
	private String createTriggerGroupSetKey(String groupName) {
		return "trigger_group:" + groupName;
	}

	/**
	 * Creates a trigger hash key in the form of: <br>
	 * 'trigger:&#60;trigger_group_name&#62;:&#60;trigger_name&#62;'
	 * 
	 * @param groupName the group name
	 * @param triggerName the trigger name
	 * @return the trigger hash key
	 */
	private String createTriggerHashKey(String groupName, String triggerName) {
		return "trigger:" + groupName + ":" + triggerName;
	}

	@Override
	public boolean removeTrigger(TriggerKey triggerKey)
			throws JobPersistenceException {
		boolean removed = false;
		String triggerHashKey = createTriggerHashKey(triggerKey.getGroup(), triggerKey.getName());
      try (Jedis jedis = pool.getResource()) {
         lockPool.acquire();
			if (jedis.exists(triggerHashKey)) {
				removeTrigger(triggerKey, jedis);
				removed = true;
			}			
		} catch (Exception ex) {
			log.error("could not remove trigger: " + triggerHashKey, ex);
			throw new JobPersistenceException(ex.getMessage(), ex.getCause());
		} finally {
         lockPool.release();
		}
		return removed;		
	}

	/**
	 * Removes the trigger from redis.
	 *
	 * @param triggerKey the trigger key
	 * jedis thread-safe redis connection
	 * @throws JobPersistenceException
	 */
	private void removeTrigger(TriggerKey triggerKey, Jedis jedis) 
			throws JobPersistenceException {
		String triggerHashKey = createTriggerHashKey(triggerKey.getGroup(), triggerKey.getName());
		String triggerGroupSetKey = createTriggerGroupSetKey(triggerKey.getGroup());
		
		jedis.srem(TRIGGERS_SET, triggerHashKey);
		jedis.srem(triggerGroupSetKey, triggerHashKey);
		
		String jobHashkey = jedis.hget(triggerHashKey, JOB_HASH_KEY);
		String jobTriggerSetkey = createJobTriggersSetKey(jobHashkey.split(":")[1], jobHashkey.split(":")[2]); 
		jedis.srem(jobTriggerSetkey, triggerHashKey);
		
		if (jedis.scard(triggerGroupSetKey) == 0) {
			jedis.srem(TRIGGER_GROUPS_SET, triggerGroupSetKey);
		}
		
		// handling orphaned jobs
		if (jedis.scard(jobTriggerSetkey) == 0 && jedis.exists(jobHashkey)) {
			if (!Boolean.parseBoolean(jedis.hget(jobHashkey, IS_DURABLE))) {
				JobKey jobKey = new JobKey(jobHashkey.split(":")[2], jobHashkey.split(":")[1]);
				removeJob(jobKey, jedis);
				signaler.notifySchedulerListenersJobDeleted(jobKey);
			}				
		}
		
		String calendarName = jedis.hget(triggerHashKey, CALENDAR_NAME);
		if (!calendarName.isEmpty()) {
			String calendarTriggersSetKey = createCalendarTriggersSetKey(calendarName);
			jedis.srem(calendarTriggersSetKey, triggerHashKey);			
		}			
			
		// removing trigger state
		unsetTriggerState(triggerHashKey);		
		jedis.del(triggerHashKey);
	}

	/**
	 * Sets a trigger state by adding the trigger to a relevant sorted set, using it's next fire time as the score.<br>
	 * The caller should handle synchronization.
	 *
	 * @param state the new state to be set
	 * @param score the trigger's next fire time
	 * @param triggerHashKey the trigger hash key
	 * @return 1- if set, 0- if the trigger is already member of the state's sorted set and it's score was updated, -1- if setting state failed
	 * @throws JobPersistenceException
	 */
	private long setTriggerState(RedisTriggerState state, double score, String triggerHashKey) throws JobPersistenceException {
		long success = -1;
      try (Jedis jedis = pool.getResource()) {
			long removed = unsetTriggerState(triggerHashKey);			
			if (state != null && removed >= 0)
				success = jedis.zadd(state.getKey(), score, triggerHashKey);			
		} catch (Exception ex) {
         log.error("could not set state " + state + " for trigger: " + triggerHashKey);
			throw new JobPersistenceException(ex.getMessage(), ex.getCause());
      }
		return success;		
	}
	
	/**
	 * Unsets a state of a given trigger key by removing the trigger from the trigger state zsets. <br>
	 * The caller should handle synchronization.
	 *
	 * @param triggerHashKey the trigger hash key
	 * @param jedis the redis client
	 * @return 1- if removed, 0- if the trigger was stateless, -1- if unset state failed  
	 * @throws JobPersistenceException 
	 */
	private long unsetTriggerState(String triggerHashKey) throws JobPersistenceException {
		Long removed = -1L;
      try (Jedis jedis = pool.getResource()) {
			Pipeline p = jedis.pipelined();
			p.zrem(RedisTriggerState.WAITING.getKey(), triggerHashKey);
			p.zrem(RedisTriggerState.PAUSED.getKey(), triggerHashKey);
			p.zrem(RedisTriggerState.BLOCKED.getKey(), triggerHashKey);
			p.zrem(RedisTriggerState.PAUSED_BLOCKED.getKey(), triggerHashKey);
			p.zrem(RedisTriggerState.ACQUIRED.getKey(), triggerHashKey);
			p.zrem(RedisTriggerState.COMPLETED.getKey(), triggerHashKey);
			p.zrem(RedisTriggerState.ERROR.getKey(), triggerHashKey);
			List<Object> results = p.syncAndReturnAll();
			
			for (Object result : results) {
				removed = (Long)result;
				if (removed == 1) {
					jedis.hset(triggerHashKey, LOCKED_BY, "");
					jedis.hset(triggerHashKey, LOCK_TIME, "");
					break;
				}
			}
		} catch (Exception ex) {
			removed = -1L;
			log.error("could not unset state for trigger: " + triggerHashKey);
			throw new JobPersistenceException(ex.getMessage(), ex.getCause());
      }
		return removed;
	}

	@Override
	public boolean removeTriggers(List<TriggerKey> triggerKeys)
			throws JobPersistenceException {
      boolean removed = triggerKeys.size() > 0;
		for (TriggerKey triggerKey : triggerKeys)
			removed = removeTrigger(triggerKey) && removed;
				
		return removed;
	}

	@Override
	public boolean replaceTrigger(TriggerKey triggerKey,
			OperableTrigger newTrigger) throws JobPersistenceException {
		boolean replaced = false;
		String triggerHashKey = createTriggerHashKey(triggerKey.getGroup(), triggerKey.getName());
      try (Jedis jedis = pool.getResource()) {
         lockPool.acquire();
			
			String jobHashKey =jedis.hget(triggerHashKey, JOB_HASH_KEY); 
			if (jobHashKey == null || jobHashKey.isEmpty())
				throw new JobPersistenceException("trigger does not exist or no job is associated with the trigger");			
			
			if (!jobHashKey.equals(createJobHashKey(newTrigger.getJobKey().getGroup(), newTrigger.getJobKey().getName())))
				throw new JobPersistenceException("the new trigger is associated with a diffrent job than the existing trigger");
			
			removeTrigger(triggerKey, jedis);
			storeTrigger(newTrigger, false, jedis);
			replaced = true;
		} catch (Exception ex) {
			log.error("could not replace trigger: " + triggerHashKey, ex);
			throw new JobPersistenceException(ex.getMessage(), ex.getCause());
		} finally {
         lockPool.release();
		}
		return replaced;
	}

	@Override
	public OperableTrigger retrieveTrigger(TriggerKey triggerKey) throws JobPersistenceException {
		OperableTrigger trigger = null;
		String triggerHashKey = createTriggerHashKey(triggerKey.getGroup(), triggerKey.getName());
      try (Jedis jedis = pool.getResource()) {
         lockPool.acquire();
			trigger = retrieveTrigger(triggerKey, jedis);			
		} catch (Exception ex) {
			log.error("could not retrieve trigger: " + triggerHashKey, ex);
			throw new JobPersistenceException(ex.getMessage(), ex.getCause());
		} finally {
         lockPool.release();
		}
		return trigger;
	}

	/**
	 * Retrieves trigger from redis.
	 *
	 * @param triggerKey the trigger key
	 * @param jedis thread-safe redis connection
	 * @return the operable trigger
	 * @throws JobPersistenceException
	 */
	private OperableTrigger retrieveTrigger(TriggerKey triggerKey, Jedis jedis) throws JobPersistenceException {
		String triggerHashKey = createTriggerHashKey(triggerKey.getGroup(), triggerKey.getName());
		Map<String, String> trigger = jedis.hgetAll(triggerHashKey);
		if (!jedis.exists(triggerHashKey)) {
			log.debug("trigger does not exist for key: " + triggerHashKey);
			return null;
		}			
		
      if (!jedis.exists(trigger.get(JOB_HASH_KEY))) {
         log.warn("job: " + trigger.get(JOB_HASH_KEY) + " does not exist for trigger: " + triggerHashKey);
         return null;
      }
		return toOperableTrigger(triggerKey, trigger);
	}

	private OperableTrigger toOperableTrigger(TriggerKey triggerKey, Map<String, String> trigger) {
		if (TRIGGER_TYPE_SIMPLE.equals(trigger.get(TRIGGER_TYPE))) {
			SimpleTriggerImpl simpleTrigger = new SimpleTriggerImpl();
			setOperableTriggerFields(triggerKey, trigger, simpleTrigger);
			if (trigger.get(REPEAT_COUNT) != null && !trigger.get(REPEAT_COUNT).isEmpty())
				simpleTrigger.setRepeatCount(Integer.parseInt(trigger.get(REPEAT_COUNT)));
			if (trigger.get(REPEAT_INTERVAL) != null && !trigger.get(REPEAT_INTERVAL).isEmpty())
				simpleTrigger.setRepeatInterval(Long.parseLong(trigger.get(REPEAT_INTERVAL)));
			if (trigger.get(TIMES_TRIGGERED) != null && !trigger.get(TIMES_TRIGGERED).isEmpty())
				simpleTrigger.setTimesTriggered(Integer.parseInt(trigger.get(TIMES_TRIGGERED)));
			
			return simpleTrigger;
		} else if (TRIGGER_TYPE_CRON.equals(trigger.get(TRIGGER_TYPE))) {
			CronTriggerImpl cronTrigger = new CronTriggerImpl();
			setOperableTriggerFields(triggerKey, trigger, cronTrigger);
			if (trigger.get(TIME_ZONE_ID) != null && !trigger.get(TIME_ZONE_ID).isEmpty())
				cronTrigger.getTimeZone().setID(trigger.get(TIME_ZONE_ID).isEmpty() ? null : trigger.get(TIME_ZONE_ID));
			try {
				if (trigger.get(CRON_EXPRESSION) != null && !trigger.get(CRON_EXPRESSION).isEmpty())
					cronTrigger.setCronExpression(trigger.get(CRON_EXPRESSION).isEmpty() ? null : trigger.get(CRON_EXPRESSION));
			} catch (ParseException ex) {
				log.warn("could not parse cron_expression: " + trigger.get(CRON_EXPRESSION) + " for trigger: " + createTriggerHashKey(triggerKey.getGroup(), triggerKey.getName()));
			}
			
			return cronTrigger;					
		} else { // other trigger types are not supported
			 throw new UnsupportedOperationException();
		}
	}

	private void setOperableTriggerFields(TriggerKey triggerKey, Map<String, String> trigger, OperableTrigger operableTrigger) {
		operableTrigger.setKey(triggerKey);
