/********************************************************************
 * Copyright © 2006 VHA. All rights reserved
 ********************************************************************/

package gov.va.med.fw.batchprocess;

import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.Validate;
import org.hibernate.LockMode;
import org.hibernate.Query;
import org.hibernate.Session;
import org.quartz.SchedulerException;
import org.springframework.scheduling.quartz.JobDetailBean;

import gov.va.med.fw.batchprocess.model.JobConfig;
import gov.va.med.fw.batchprocess.model.JobResult;
import gov.va.med.fw.batchprocess.model.JobStatus;
import gov.va.med.fw.model.EntityKey;
import gov.va.med.fw.persistent.DAOException;
import gov.va.med.fw.persistent.DAOOperations;
import gov.va.med.fw.persistent.hibernate.AbstractDAOAction;
import gov.va.med.fw.scheduling.AuditableScheduledProcess;
import gov.va.med.fw.scheduling.ScheduledProcessInvocationContext;
import gov.va.med.fw.scheduling.ScheduledProcessTriggerEvent;
import gov.va.med.fw.scheduling.SchedulingService;
import gov.va.med.fw.security.LoginManager;
import gov.va.med.fw.service.AbstractComponent;
import gov.va.med.fw.service.ServiceException;
import gov.va.med.fw.util.Reflector;
import gov.va.med.fw.util.SystemUtils;

/**
 * Initial javadoc for class BatchProcessServiceImpl. 
 * 
 * Created Sep 29, 2006 1:48:33 PM
 * 
 * @author DNS   BOHMEG
 */
public class BatchProcessServiceImpl extends AbstractComponent implements
		BatchProcessService {
	private static String NAMED_QUERY_GET_JOB_RESULTS_BY_STATUS = "jobResultQuery_GetByStatus";

	private static String NAMED_QUERY_GET_JOB_RESULTS_BY_JOB_NAME_AND_STATUS = "jobResultQuery_GetFinishedByJobNameAndStatus";

	private static String NAMED_QUERY_GET_JOB_RESULTS_BY_JOB_NAME_AND_JOB_GROUP_AND_STATUS = "jobResultQuery_GetFinishedByJobNameAndJobGroupAndStatus";

	private static String NAMED_QUERY_GET_JOB_CONFIG_BY_JOB_NAME = "jobConfigQuery_GetByJobName";

	private static String NAMED_QUERY_GET_JOB_CONFIG_BY_JOB_NAME_AND_JOB_GROUP = "jobConfigQuery_GetByJobNameAndJobGroup";

	private static String PARAM_STATUS = "status";

	private static String PARAM_JOB_NAME = "jobName";

	private static String PARAM_JOB_GROUP = "jobGroup";

	private DAOOperations dao;

	private LoginManager loginManager;
	
	private SchedulingService schedulingService;
	
	/*
	 * (non-Javadoc)
	 * 
	 * @see gov.va.med.fw.batchprocess.BatchProcessService#saveJobResult(gov.va.med.fw.batchprocess.model.JobResult)
	 */
	public void saveJobResult(JobResult jr) throws ServiceException {
		try {
			dao.saveObject(jr);
		} catch (DAOException e) {
			throw new ServiceException("Unable to save JobResult", e);
		}
	}

	public JobResult getJobResult(EntityKey key) throws ServiceException {
		try {
			return (JobResult) dao.getByKey(key);
		} catch (DAOException e) {
			throw new ServiceException("Unable to get JobResult", e);
		}
	}

	public JobResultData saveAndOverlayJobResult(EntityKey jobResultKey,
			ProcessStatistics stats, int totalDataCount)
			throws ServiceException {
		JobResult jr = getAndLockJobResult(jobResultKey);
		// create new ProcessStatistics
		ProcessStatistics updatedStats = null;
		try {
			// create new ProcessStatistics
			updatedStats = (ProcessStatistics) stats.getClass().newInstance();

			updatedStats.importFromCSV(jr.getStatistics()); // seed it with db
															// stats
			// now overlay worker stats
			updatedStats.overlayStats(stats);

			// update status
			if (updatedStats.getNumberOfTotalRecords() >= totalDataCount) {
				updatedStats.setProcessingEndDate(new Date());
				jr.setEndDate(updatedStats.getProcessingEndDate());
				if (updatedStats.getNumberOfErrorRecords() > 0)
					jr.setStatus(JobStatus.COMPLETE_WITH_ERROR);
				else
					jr.setStatus(JobStatus.COMPLETE);
			}
			jr.setStatistics(updatedStats.exportAsCSV());

			saveJobResult(jr);
		} catch (Exception e) {
			throw new ServiceException("Unable to saveAndOverlayJobResult", e);
		}
		return new JobResultData(jr, updatedStats);
	}

	private JobResult getAndLockJobResult(EntityKey key)
			throws ServiceException {
		try {
			Map contextData = new HashMap();
			contextData.put("entityKey", key);
			AbstractDAOAction callback = new AbstractDAOAction(contextData) {
				public Object execute(Session session) {
					EntityKey key = (EntityKey) getContextData().get(
							"entityKey");
					return session.load(JobResult.class, key.getKeyValue(),
							LockMode.UPGRADE);
				}
			};
			return (JobResult) dao.execute(callback);
		} catch (DAOException e) {
			throw new ServiceException("Unable to getAndLockJobResult for key="
					+ key, e);
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see gov.va.med.fw.batchprocess.BatchProcessService#saveJobConfig(gov.va.med.fw.batchprocess.model.JobConfig)
	 */
	public void saveJobConfig(JobConfig jc) throws ServiceException {
		try {
			dao.saveObject(jc);
		} catch (DAOException e) {
			throw new ServiceException("Unable to save JobConfig", e);
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see gov.va.med.fw.batchprocess.BatchProcessService#getJobConfig(java.lang.String)
	 */
	public JobConfig getJobConfig(String jobName, String jobGroup)
			throws ServiceException {
		try {
			List data = null;
			if (jobGroup == null)
				data = dao.findByNamedQueryAndNamedParam(
						NAMED_QUERY_GET_JOB_CONFIG_BY_JOB_NAME, PARAM_JOB_NAME,
						jobName);
			else
				data = dao.findByNamedQueryAndNamedParam(
						NAMED_QUERY_GET_JOB_CONFIG_BY_JOB_NAME_AND_JOB_GROUP,
						new String[] { PARAM_JOB_NAME, PARAM_JOB_GROUP },
						new Object[] { jobName, jobGroup });
			return (data != null && !data.isEmpty()) ? (JobConfig) data.get(0)
					: null;
		} catch (DAOException e) {
			throw new ServiceException("Unable to get JobConfig for jobName: "
					+ jobName, e);
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see gov.va.med.fw.batchprocess.BatchProcessService#getJobResults(gov.va.med.fw.batchprocess.model.JobStatus)
	 */
	public List getJobResults(JobStatus status) throws ServiceException {
		try {
			return dao.findByNamedQueryAndNamedParam(
					NAMED_QUERY_GET_JOB_RESULTS_BY_STATUS, PARAM_STATUS, status
							.getCode());
		} catch (DAOException e) {
			throw new ServiceException("Unable to getJobResults with status: "
					+ status, e);
		}
	}

	/** Currently constrained to last 25 */
	public List getFinishedJobResults(String jobName, String jobGroup)
			throws ServiceException {
		try {
			Map contextData = new HashMap();
			contextData.put("jobName", jobName);
			if (jobGroup != null)
				contextData.put("jobGroup", jobGroup);

			AbstractDAOAction callback = new AbstractDAOAction(contextData) {
				public Object execute(Session session) {
					Query query = null;
					if (!getContextData().containsKey("jobGroup"))
						query = session
								.getNamedQuery(NAMED_QUERY_GET_JOB_RESULTS_BY_JOB_NAME_AND_STATUS);
					else {
						query = session
								.getNamedQuery(NAMED_QUERY_GET_JOB_RESULTS_BY_JOB_NAME_AND_JOB_GROUP_AND_STATUS);
						query.setString(PARAM_JOB_GROUP,
								(String) getContextData().get("jobGroup"));
					}
					query.setString(PARAM_JOB_NAME, (String) getContextData()
							.get("jobName"));

					query.setMaxResults(25);
					return query.list();
				}
			};
			return dao.executeFind(callback);
		} catch (DAOException e) {
			throw new ServiceException(
					"Unable to getFinishedJobResults for jobName: " + jobName,
					e);
		}
	}

	public void executeJob(ScheduledProcessTriggerEvent triggerEvent) throws ServiceException {
		executeJob((ScheduledProcessInvocationContext) triggerEvent.getPayload());
	}
	
	/*
	 * (non-Javadoc)
	 * 
	 * @see gov.va.med.fw.scheduling.SchedulingService#fireJob(gov.va.med.fw.scheduling.ScheduledProcessInvocationContext)
	 */
	public void executeJob(ScheduledProcessInvocationContext invocationContext)
			throws ServiceException {
		if(logger.isInfoEnabled()) {
			logger.info("Executing job/process [" + invocationContext.getJobName() + "] for executionContext/user [" + invocationContext.getExecutionContext() + "]");
		}
		
		String auditInfo = null;
		long beforeUsedMemory = SystemUtils.getUsedMemory();
		try {			
			/* this implementation supports simple invocation via a Spring JobDetailName or from the actual service name
			 * and method name.  The point here is we need the serviceName and methodName and there is more than one
			 * way to get it.
			 */
			String serviceName = null;
			String methodName = null;
			if(invocationContext.getJobDetailBeanName() != null) {
				JobDetailBean job = (JobDetailBean) this.getApplicationContext()
						.getBean(invocationContext.getJobDetailBeanName(),
								JobDetailBean.class);
				serviceName = (String) job.getJobDataMap()
						.get("serviceName");
				methodName = (String) job.getJobDataMap().get("methodName");
			} else {
				serviceName = invocationContext.getServiceName();
				methodName = invocationContext.getMethodName();
			}
			Object service = this.getApplicationContext().getBean(serviceName);

			auditInfo = service instanceof AuditableScheduledProcess ? ((AuditableScheduledProcess) service).getAuditInfo(invocationContext) : serviceName + "." + methodName;
			getLoginManager().loginAnonymous(auditInfo);
			
			// see if service supports invocation via a
			// ScheduledProcessInvocationContext
			Object[] args = new Object[] { invocationContext };
			try {
				Reflector.findMethod(service, methodName, args);
			} catch (NoSuchMethodException e) {
				// no "standard" method for ScheduledProcessInvocationContext
				// ....just use wrapped arguments
				Object obj = invocationContext.getInvocationArguments();	
				args = obj instanceof Object[] ? (Object[]) obj : new Object[] {obj};
			}

			Reflector.invoke(service, methodName, args);
		} catch (Throwable e) {
			throw new ServiceException("Unable to execute job with ScheduledProcessInvocationContext: " + invocationContext, e);
		} finally {
			getLoginManager().logout();
			
			SystemUtils.logMemory(true); // note this will suggest GC			
			SystemUtils.logMemoryDelta(auditInfo, beforeUsedMemory);			
		}
	}

	/**
	 * @return Returns the dao.
	 */
	public DAOOperations getDao() {
		return dao;
	}

	/**
	 * @param dao
	 *            The dao to set.
	 */
	public void setDao(DAOOperations dao) {
		this.dao = dao;
	}

	public void afterPropertiesSet() {
		Validate.notNull(dao, "A dao is required");
		Validate.notNull(loginManager, "A loginManager is required");
		Validate.notNull(schedulingService, "A schedulingService is required");
	}

	/**
	 * @return Returns the loginManager.
	 */
	public LoginManager getLoginManager() {
		return loginManager;
	}

	/**
	 * @param loginManager The loginManager to set.
	 */
	public void setLoginManager(LoginManager loginManager) {
		this.loginManager = loginManager;
	}

	/* (non-Javadoc)
	 * @see gov.va.med.fw.batchprocess.BatchProcessService#cancelActiveJob(gov.va.med.fw.model.EntityKey)
	 */
	public void cancelActiveJob(EntityKey key) throws ServiceException {
		/* some jobs are sync and some are async - to cover these scenarios, must cancel the job
		 * in 2 different ways
		 */

		JobResult jobResult = this.getJobResult(key);
		if(JobStatus.IN_PROCESS.getCode().equals(jobResult.getStatus().getCode())) {
			// do this first
			jobResult.setStatus(JobStatus.CANCELLED);
			saveJobResult(jobResult);
			
			// do this second
			try {
				schedulingService.interrupt(jobResult.getJobName(), jobResult.getJobGroup());
			} catch(SchedulerException e) {
				throw new ServiceException("Unable to interrupt the job during cancel", e);
			}			
		}
	}

	/* (non-Javadoc)
	 * @see gov.va.med.fw.batchprocess.BatchProcessService#errActiveJob(gov.va.med.fw.model.EntityKey)
	 */
	public void errActiveJob(EntityKey key) throws ServiceException {
		JobResult jobResult = this.getJobResult(key);
		if(JobStatus.IN_PROCESS.getCode().equals(jobResult.getStatus().getCode())) {
			jobResult.setStatus(JobStatus.ERROR);
			saveJobResult(jobResult);
		}
	}

	/**
	 * @return Returns the schedulingService.
	 */
	public SchedulingService getSchedulingService() {
		return schedulingService;
	}

	/**
	 * @param schedulingService The schedulingService to set.
	 */
	public void setSchedulingService(SchedulingService schedulingService) {
		this.schedulingService = schedulingService;
	}
}
