/*
 * Copyright 2012 LinkedIn, Inc
 * 
 * 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 java.io.IOException;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import org.apache.commons.dbutils.DbUtils;
import org.apache.commons.dbutils.QueryRunner;
import org.apache.commons.dbutils.ResultSetHandler;
import org.apache.log4j.Logger;

import org.joda.time.DateTimeZone;
import org.joda.time.ReadablePeriod;

import azkaban.database.AbstractJdbcLoader;
import azkaban.utils.GZIPUtils;
import azkaban.utils.JSONUtils;
import azkaban.utils.Props;


public class JdbcScheduleLoader extends AbstractJdbcLoader implements ScheduleLoader {
	private static Logger logger = Logger.getLogger(JdbcScheduleLoader.class);

	private EncodingType defaultEncodingType = EncodingType.GZIP;
	
	private static final String scheduleTableName = "schedules";

	private static String SELECT_ALL_SCHEDULES =
			"SELECT schedule_id, project_id, project_name, flow_name, status, first_sched_time, timezone, period, last_modify_time, next_exec_time, submit_time, submit_user, enc_type, schedule_options FROM " + scheduleTableName;
	
	private static String INSERT_SCHEDULE = 
			"INSERT INTO " + scheduleTableName + " ( project_id, project_name, flow_name, status, first_sched_time, timezone, period, last_modify_time, next_exec_time, submit_time, submit_user, enc_type, schedule_options) values (?,?,?,?,?,?,?,?,?,?,?,?,?)";
	
	private static String REMOVE_SCHEDULE_BY_KEY = 
			"DELETE FROM " + scheduleTableName + " WHERE schedule_id=?";
	
	private static String UPDATE_SCHEDULE_BY_KEY = 
			"UPDATE " + scheduleTableName + " SET status=?, first_sched_time=?, timezone=?, period=?, last_modify_time=?, next_exec_time=?, submit_time=?, submit_user=?, enc_type=?, schedule_options=? WHERE schedule_id=?";
	
	private static String UPDATE_NEXT_EXEC_TIME = 
			"UPDATE " + scheduleTableName + " SET next_exec_time=? WHERE schedule_id=?";

	public EncodingType getDefaultEncodingType() {
		return defaultEncodingType;
	}

	public void setDefaultEncodingType(EncodingType defaultEncodingType) {
		this.defaultEncodingType = defaultEncodingType;
	}
	
	public JdbcScheduleLoader(Props props) {
		super(props);
	}

	@Override
	public List<Schedule> loadSchedules() throws ScheduleManagerException {
		logger.info("Loading all schedules from db.");
		Connection connection = getConnection();

		QueryRunner runner = new QueryRunner();
		ResultSetHandler<List<Schedule>> handler = new ScheduleResultHandler();
	
		List<Schedule> schedules;
		
		try {
			schedules = runner.query(connection, SELECT_ALL_SCHEDULES, handler);
		} catch (SQLException e) {
			logger.error(SELECT_ALL_SCHEDULES + " failed.");

			DbUtils.closeQuietly(connection);
			throw new ScheduleManagerException("Loading schedules from db failed. ", e);
		} finally {
			DbUtils.closeQuietly(connection);
		}
		
		logger.info("Now trying to update the schedules");
		
		// filter the schedules
		for(Schedule sched : schedules) {
			if(!sched.updateTime()) {
				logger.info("Schedule " + sched.getScheduleName() + " was scheduled before azkaban start, skipping it.");
				schedules.remove(sched);
				removeSchedule(sched);
			}
			else {
				logger.info("Recurring schedule, need to update next exec time");
				try {
					updateNextExecTime(sched);
				} catch (Exception e) {
					e.printStackTrace();
					throw new ScheduleManagerException("Update next execution time failed.", e);
				} 
				logger.info("Schedule " + sched.getScheduleName() + " loaded and updated.");
			}
		}
		
		
				
		logger.info("Loaded " + schedules.size() + " schedules.");
		
		return schedules;
	}

	@Override
	public void removeSchedule(Schedule s) throws ScheduleManagerException {		
		logger.info("Removing schedule " + s.getScheduleName() + " from db.");

		QueryRunner runner = createQueryRunner();
		try {
			int removes =  runner.update(REMOVE_SCHEDULE_BY_KEY, s.getScheduleId());
			if (removes == 0) {
				throw new ScheduleManagerException("No schedule has been removed.");
			}
		} catch (SQLException e) {
			logger.error(REMOVE_SCHEDULE_BY_KEY + " failed.");
			throw new ScheduleManagerException("Remove schedule " + s.getScheduleName() + " from db failed. ", e);
		}
	}
	
	
	public void insertSchedule(Schedule s) throws ScheduleManagerException {
		logger.info("Inserting schedule " + s.getScheduleName() + " into db.");
		insertSchedule(s, defaultEncodingType);
	}

	public void insertSchedule(Schedule s, EncodingType encType) throws ScheduleManagerException {
		
		String json = JSONUtils.toJSON(s.optionsToObject());
		byte[] data = null;
		try {
			byte[] stringData = json.getBytes("UTF-8");
			data = stringData;
	
			if (encType == EncodingType.GZIP) {
				data = GZIPUtils.gzipBytes(stringData);
			}
			logger.debug("NumChars: " + json.length() + " UTF-8:" + stringData.length + " Gzip:"+ data.length);
		}
		catch (IOException e) {
			throw new ScheduleManagerException("Error encoding the schedule options. " + s.getScheduleName());
		}
		
		QueryRunner runner = createQueryRunner();
		try {
			int inserts =  runner.update( 
					INSERT_SCHEDULE, 
					s.getProjectId(),
					s.getProjectName(),
					s.getFlowName(), 
					s.getStatus(), 
					s.getFirstSchedTime(), 
					s.getTimezone().getID(), 
					Schedule.createPeriodString(s.getPeriod()), 
					s.getLastModifyTime(), 
					s.getNextExecTime(), 
					s.getSubmitTime(), 
					s.getSubmitUser(),
					encType.getNumVal(),
					data);
			
			long id = runner.query(LastInsertID.LAST_INSERT_ID, new LastInsertID());

			if (id == -1l) {
				throw new ScheduleManagerException("Execution id is not properly created.");
			}
			logger.info("Schedule given " + s.getScheduleIdentityPair() + " given id " + id);
			s.setScheduleId((int)id);
			
			if (inserts == 0) {
				throw new ScheduleManagerException("No schedule has been inserted.");
			}
		} catch (SQLException e) {
			logger.error(INSERT_SCHEDULE + " failed.");
			throw new ScheduleManagerException("Insert schedule " + s.getScheduleName() + " into db failed. ", e);
		}
	}
	
	@Override
	public void updateNextExecTime(Schedule s) throws ScheduleManagerException 
	{
		logger.info("Update schedule " + s.getScheduleName() + " into db. ");
		Connection connection = getConnection();
		QueryRunner runner = new QueryRunner();
		try {
			
			runner.update(connection, UPDATE_NEXT_EXEC_TIME, s.getNextExecTime(), s.getScheduleId()); 
		} catch (SQLException e) {
			e.printStackTrace();
			logger.error(UPDATE_NEXT_EXEC_TIME + " failed.", e);
			throw new ScheduleManagerException("Update schedule " + s.getScheduleName() + " into db failed. ", e);
		} finally {
			DbUtils.closeQuietly(connection);
		}
	}
	
	@Override
	public void updateSchedule(Schedule s) throws ScheduleManagerException {
		logger.info("Updating schedule " + s.getScheduleName() + " into db.");
		updateSchedule(s, defaultEncodingType);
	}
		
	public void updateSchedule(Schedule s, EncodingType encType) throws ScheduleManagerException {

		String json = JSONUtils.toJSON(s.optionsToObject());
		byte[] data = null;
		try {
			byte[] stringData = json.getBytes("UTF-8");
			data = stringData;
	
			if (encType == EncodingType.GZIP) {
				data = GZIPUtils.gzipBytes(stringData);
			}
			logger.debug("NumChars: " + json.length() + " UTF-8:" + stringData.length + " Gzip:"+ data.length);
		}
		catch (IOException e) {
			throw new ScheduleManagerException("Error encoding the schedule options " + s.getScheduleName());
		}

		QueryRunner runner = createQueryRunner();
	
		try {
			int updates =  runner.update( 
					UPDATE_SCHEDULE_BY_KEY, 
					s.getStatus(), 
					s.getFirstSchedTime(), 
					s.getTimezone().getID(), 
					Schedule.createPeriodString(s.getPeriod()), 
					s.getLastModifyTime(), 
					s.getNextExecTime(), 
					s.getSubmitTime(), 
					s.getSubmitUser(), 	
					encType.getNumVal(),
					data,
					s.getScheduleId());
			if (updates == 0) {
				throw new ScheduleManagerException("No schedule has been updated.");
			}
		} catch (SQLException e) {
			logger.error(UPDATE_SCHEDULE_BY_KEY + " failed.");
			throw new ScheduleManagerException("Update schedule " + s.getScheduleName() + " into db failed. ", e);
		}
	}

	
	private static class LastInsertID implements ResultSetHandler<Long> {
		private static String LAST_INSERT_ID = "SELECT LAST_INSERT_ID()";
		
		@Override
		public Long handle(ResultSet rs) throws SQLException {
			if (!rs.next()) {
				return -1l;
			}

			long id = rs.getLong(1);
			return id;
		}
		
	}
	
	public class ScheduleResultHandler implements ResultSetHandler<List<Schedule>> {
		@Override
		public List<Schedule> handle(ResultSet rs) throws SQLException {
			if (!rs.next()) {
				return Collections.<Schedule>emptyList();
			}
			
			ArrayList<Schedule> schedules = new ArrayList<Schedule>();
			do {
				int scheduleId = rs.getInt(1);
				int projectId = rs.getInt(2);
				String projectName = rs.getString(3);
				String flowName = rs.getString(4);
				String status = rs.getString(5);
				long firstSchedTime = rs.getLong(6);
				DateTimeZone timezone = DateTimeZone.forID(rs.getString(7));
				ReadablePeriod period = Schedule.parsePeriodString(rs.getString(8));
				long lastModifyTime = rs.getLong(9);
				long nextExecTime = rs.getLong(10);
				long submitTime = rs.getLong(11);
				String submitUser = rs.getString(12);
				int encodingType = rs.getInt(13);
				byte[] data = rs.getBytes(14);
				
				Object optsObj = null;
				if (data != null) {
					EncodingType encType = EncodingType.fromInteger(encodingType);

					try {
						// Convoluted way to inflate strings. Should find common package or helper function.
						if (encType == EncodingType.GZIP) {
							// Decompress the sucker.
							String jsonString = GZIPUtils.unGzipString(data, "UTF-8");
							optsObj = JSONUtils.parseJSONFromString(jsonString);
						}
						else {
							String jsonString = new String(data, "UTF-8");
							optsObj = JSONUtils.parseJSONFromString(jsonString);
						}	
					} catch (IOException e) {
						throw new SQLException("Error reconstructing schedule options " + projectName + "." + flowName);
					}
				}
				
				Schedule s = new Schedule(scheduleId, projectId, projectName, flowName, status, firstSchedTime, timezone, period, lastModifyTime, nextExecTime, submitTime, submitUser);
				if (optsObj != null) {
					s.createAndSetScheduleOptions(optsObj);
				}
				
				schedules.add(s);
			} while (rs.next());
			
			return schedules;
		}
		
	}
	
	private Connection getConnection() throws ScheduleManagerException {
		Connection connection = null;
		try {
			connection = super.getDBConnection(false);
		} catch (Exception e) {
			DbUtils.closeQuietly(connection);
			throw new ScheduleManagerException("Error getting DB connection.", e);
		}
		
		return connection;
	}
}