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

package gov.va.med.fw.batchprocess;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Date;
import java.util.Iterator;
import java.util.List;

import org.apache.commons.lang.StringUtils;
import org.quartz.Scheduler;

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.scheduling.AbstractScheduledProcess;
import gov.va.med.fw.scheduling.ScheduledProcessInvocationContext;
import gov.va.med.fw.service.ServiceException;

/**
 * Abstract base class for batch processing of bulk data.  Data can be acquired
 * in any manner that sublasses choose via the acquireData implementation.
 * 
 * <p>Tracks in process statistics via JobResult instance.  Allows for archival of
 * process statistics.
 * 
 * <p>Implementations should strive to exhibit the following characteristics:
 * <ul>
 * 	<li>Efficent - Perform as efficient as possible against a transactionally modeled database 
 * 	<li>Re-execution safe - The process can be executed multiple times in a row without any error
 * 		to normal business processes (eg, business requirement stating job can not be run more than once a day)
 * 	<li>Re-execution efficient - Subsequent executions should not process the same records that were previously processed.
 * 		This could be considered a non-idempotent process in that it will not yield the same result each time it is
 * 		executed (due to less records being processed). 
 * 	<li>Cancellable - Every process can be cancelled on demand via the UI 
 * 	<li>Incremental - Via execute args, the UI can throttle the number of records processed
 * </ul>
 * 
 * Created Feb 1, 2006 3:28:37 PM
 * 
 * @author DNS   BOHMEG
 */
public abstract class AbstractDataProcess extends AbstractScheduledProcess {
	public static final int DEFAULT_JOB_RESULT_UPDATE_INTERVAL = 5;
	private static final String JOB_RESULT_ENTITY = "jobResultEntity";
	private static final String JOB_NAME = "jobName";
	private static final String JOB_GROUP_NAME = "jobGroupName";
	
	private BatchProcessService batchProcessService;
	private String processName;
	private DataProcessCompletedHandler dataProcessCompletedHandler;
	private ProcessStatisticsHandler processStatisticsHandler;
	private boolean allowConcurrentProcessExecution = false;

	protected final void executeProcess(ScheduledProcessInvocationContext invocationContext) throws Exception {
		DataProcessExecutionContext context = createDataProcessExecutionContext();
		String jobName = getProcessName(); // used for JobResult/JobConfig persistence (want it to be the originating JobDetail name)
		String jobGroup = Scheduler.DEFAULT_GROUP;
		if (invocationContext != null) {
			/* Allow the invocationContext to rely on default group and name */
			if (invocationContext.getJobGroup() != null)
				jobGroup = invocationContext.getJobGroup();
			if (invocationContext.getJobName() != null)
				jobName = invocationContext.getJobName();
			context.setExecutionArguments(invocationContext
					.getInvocationArguments());
			context.setInitiater(invocationContext.getExecutionContext());
		}
		context.getContextData().put(JOB_NAME, jobName);
		context.getContextData().put(JOB_GROUP_NAME, jobGroup);
		
		ProcessStatistics stats = context.getProcessStatistics();		
		stats.setProcessingStartDate(new Date());
		stats.setProcessName(getProcessName());
		try {
			stats.setExecutedOnServer(InetAddress.getLocalHost().getHostName());
		} catch(UnknownHostException e) {
			// oh well, we tried
			stats.setExecutedOnServer("Unknown");
		}		
		
		boolean proceed = true;
		JobStatus jobStatus = JobStatus.IN_PROCESS;
		if(!allowConcurrentProcessExecution) {
			if(containsInProcessStatistics(context)) {
				jobStatus = JobStatus.NOT_EXECUTED_SINCE_INFLIGHT_PROCESS;
				proceed = false;
			}
		}
		// create a JobResult for persistent tracking
		JobResult jobResult = createJobResult(context, jobStatus);
		
		if(!proceed) {
			if(logger.isInfoEnabled()) {
				logger.info("Not executing process [" + jobName + "] since there is already one with IN_PROCESS statistics");
			}
			return;
		}
		
		boolean error = false;
		try {
			executeProcess(context);
		} catch(Exception e) {
			error = true;			
			if(logger.isErrorEnabled())
				logger.error("AbstractDataProcess.executeProcess threw exception with message: " + e.getMessage(), e);
			throw e;
		} finally {
			if(ownsJobCompletion(context) || error) {
				stats.setProcessingEndDate(new Date());
				if(this.isInterrupted(context)) {
					// must re-retrieve the JobResult since it was updated externally (avoid StaleObjectException)
					jobResult = this.batchProcessService.getJobResult(jobResult.getEntityKey());
				}
				jobResult.setEndDate(stats.getProcessingEndDate());
				
				// update jobResult with final results (should also have been incremental updates by the subclasses)
				if(error) {
					jobResult.setStatus(JobStatus.ERROR);
				} else if(isInterrupted(context)) {
					jobResult.setStatus(JobStatus.CANCELLED);
				} else if(stats.containsErrors()) {
					jobResult.setStatus(JobStatus.COMPLETE_WITH_ERROR);					
				} else {
					jobResult.setStatus(JobStatus.COMPLETE);
				}
				
				jobResult.setStatistics(stats.exportAsCSV());
				batchProcessService.saveJobResult(jobResult);				
				
				processStatistics(context);				
			}
		}		
	}
	
	private final boolean containsInProcessStatistics(DataProcessExecutionContext context) throws ServiceException {
		boolean foundInProcess = false;
		List activeJobs = batchProcessService.getJobResults(JobStatus.IN_PROCESS);
		Iterator itr = activeJobs != null ? activeJobs.iterator() : null;
		JobResult inProcessResult = null; 
		while(itr != null && itr.hasNext()) {
			inProcessResult = (JobResult) itr.next();
			if(inProcessResult.getJobName().equals(getJobName(context))) {
				foundInProcess = true;
				break;
			}
		}
		return foundInProcess;
	}
	
	/** not final for framework overrides */
	protected boolean ownsJobCompletion(DataProcessExecutionContext context) {
		return true;
	}
	
	
	protected final void updateJobResult(DataProcessExecutionContext context, JobResult jobResult) throws Exception {		
		if(this.isInterrupted(context)) {
			if(logger.isWarnEnabled())
				logger.warn("Prior to updating the JobResult, it was discovered that [" + jobResult.getJobName() +
						"] has been canceled...deferring current JobResult update until the job winds itself out of interrupted status");
		} else {		
			batchProcessService.saveJobResult(jobResult);
		}
	}

	protected final void updateJobResult(DataProcessExecutionContext context) {
		JobResult jobResult = this.getJobResult(context);		
		try {
			jobResult.setStatistics(context.getProcessStatistics().exportAsCSV());
			updateJobResult(context, jobResult);
		} catch(Exception e) {
			this.throwIllegalStateException("Unable to update JobResult", e);
		}
	}
	
	private JobResult createJobResult(DataProcessExecutionContext context, JobStatus jobStatus) throws Exception {
		JobResult jobResult = new JobResult();
		jobResult.setStartDate(context.getProcessStatistics().getProcessingStartDate());
		jobResult.setJobName(getJobName(context));
		jobResult.setJobGroup(getJobGroupName(context));
		jobResult.setStatistics(context.getProcessStatistics().exportAsCSV());
		jobResult.setStatus(jobStatus);
		jobResult.setContext(context.getInitiater());
		batchProcessService.saveJobResult(jobResult);
		context.getContextData().put(JOB_RESULT_ENTITY, jobResult);
		return jobResult;
	}
	
	protected JobResult getJobResult(DataProcessExecutionContext context) {
		return (JobResult) context.getContextData().get(JOB_RESULT_ENTITY);
	}
	
	// not marked final here for framework overrides
	protected void executeProcess(DataProcessExecutionContext context) throws Exception {
		List data = acquireData(context);		
		if (data != null && !data.isEmpty()) {
			processData(context, data);
		}							
		handleDataProcessCompleted(context);
	}
	
	/**
	 * @return Returns the interrupted.
	 */
	public boolean isInterrupted(DataProcessExecutionContext context) {
		if(context.isInterrupted())
			return true;
		
		/* Interruption can either be by someone 1) calling interrupt on this instance or
		 * 2) someone setting the JobResult in the database to be JobResult.CANCELLED
		 * 
		 * we must check both here
		 */ 
		if(super.isInterrupted()) {
			context.setInterrupted(true);
			return true;
		}
		
		// need to recheck the database
		try {
			JobResult jobResult = this.getBatchProcessService().getJobResult(getJobResult(context).getEntityKey());
			if(jobResult != null) {
				boolean interrupted = jobResult.getStatus().getCode().equals(JobStatus.CANCELLED.getCode());
				if(interrupted)
					context.setInterrupted(true);
				return interrupted;
			}
			
			return false;
		} catch(ServiceException e) {
			RuntimeException e2 = new IllegalStateException("Unable to retrieve the JobResult");
			e2.initCause(e);
			throw e2;
		}
	}
	
		
	protected void handleDataProcessCompleted(DataProcessExecutionContext context) {
		if (logger.isErrorEnabled() && !context.getExceptionData().isEmpty()) {
			logger.error("Data Processing resulted in exceptions for data: " +
					context.getExceptionData());
		}
		isInterrupted(context);
		if(dataProcessCompletedHandler != null)
			dataProcessCompletedHandler.dataProcessingComplete(context);
		context.getExceptionData().clear();
		context.getProcessedData().clear();		
	}

	protected abstract void processData(DataProcessExecutionContext context, List acquiredData);

	protected abstract List acquireData(DataProcessExecutionContext context) throws Exception;

	protected final void throwIllegalStateException(String message, Throwable t) {
		RuntimeException e = new IllegalStateException(message);
		e.initCause(t);
		throw e;
	}

	protected DataProcessExecutionContext createDataProcessExecutionContext() {
		DataProcessExecutionContext context = new DataProcessExecutionContext();
		context.setProcessStatistics(createProcessStatistics());
		return context;
	}
	
	protected ProcessStatistics createProcessStatistics() {
		return new ProcessStatistics();
	}
	
	
	protected String getJobName(DataProcessExecutionContext context) {
		return (String) context.getContextData().get(JOB_NAME);		
	}

	protected String getJobGroupName(DataProcessExecutionContext context) {
		return (String) context.getContextData().get(JOB_GROUP_NAME);		
	}
	
	protected final void processStatistics(DataProcessExecutionContext context) throws Exception {
		ProcessStatistics stats = context.getProcessStatistics();
		if(isInterrupted(context))
			stats.setWasInterrupted(true);
		if(logger.isInfoEnabled())
			logger.info("ProcessStatistics [" + stats + "]");			
		if(processStatisticsHandler != null) {
			JobConfig config = this.batchProcessService.getJobConfig(getJobName(context),
					getJobGroupName(context));
			if(config != null && StringUtils.isNotBlank(config.getEmailDistributionList())) {
				processStatisticsHandler.processStatisticsWithDynamicRecipients(stats, config.getEmailsAsList());
			} else {
				processStatisticsHandler.processStatistics(stats);
			}
		}
	}

	/**
	 * @return Returns the processStatisticsHandler.
	 */
	public ProcessStatisticsHandler getProcessStatisticsHandler() {
		return processStatisticsHandler;
	}

	/**
	 * @param processStatisticsHandler The processStatisticsHandler to set.
	 */
	public void setProcessStatisticsHandler(
			ProcessStatisticsHandler processStatisticsHandler) {
		this.processStatisticsHandler = processStatisticsHandler;
	}

	
	public String getProcessName() {
		return processName != null ? processName : getAuditId();
	}

	/**
	 * @param processName The processName to set.
	 */
	public void setProcessName(String processName) {
		this.processName = processName;
	}

	/**
	 * @return Returns the dataProcessCompletedHandler.
	 */
	public DataProcessCompletedHandler getDataProcessCompletedHandler() {
		return dataProcessCompletedHandler;
	}

	/**
	 * @param dataProcessCompletedHandler The dataProcessCompletedHandler to set.
	 */
	public void setDataProcessCompletedHandler(
			DataProcessCompletedHandler dataProcessCompletedHandler) {
		this.dataProcessCompletedHandler = dataProcessCompletedHandler;
	}

	/**
	 * @return Returns the batchProcessService.
	 */
	public BatchProcessService getBatchProcessService() {
		return batchProcessService;
	}

	/**
	 * @param batchProcessService The batchProcessService to set.
	 */
	public void setBatchProcessService(BatchProcessService batchProcessService) {
		this.batchProcessService = batchProcessService;
	}

	public boolean isAllowConcurrentProcessExecution() {
		return allowConcurrentProcessExecution;
	}

	public void setAllowConcurrentProcessExecution(
			boolean allowConcurrentProcessExecution) {
		this.allowConcurrentProcessExecution = allowConcurrentProcessExecution;
	}
}
