package gov.va.med.esr.common.batchprocess;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.io.Writer;
import java.math.BigDecimal;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.GregorianCalendar;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import gov.va.med.esr.common.infra.ImpreciseDate;
import gov.va.med.esr.common.model.CommonEntityKeyFactory;
import gov.va.med.esr.common.model.comms.Correspondence;
import gov.va.med.esr.common.model.ee.MECPeriod;
import gov.va.med.esr.common.model.lookup.ComLetterTemplateType;
import gov.va.med.esr.common.model.lookup.CorrespondenceStatus;
import gov.va.med.esr.common.model.lookup.EnrollmentStatus;
import gov.va.med.esr.common.model.lookup.NameType;
import gov.va.med.esr.common.model.lookup.SSNType;
import gov.va.med.esr.common.model.messaging.IrsTransmissionLogDetailEntry;
import gov.va.med.esr.common.model.party.Address;
import gov.va.med.esr.common.model.person.BirthRecord;
import gov.va.med.esr.common.model.person.Name;
import gov.va.med.esr.common.model.person.Person;
import gov.va.med.esr.common.model.person.PersonTraits;
import gov.va.med.esr.common.model.person.SSN;
import gov.va.med.esr.common.model.person.id.PersonIdEntityKey;
import gov.va.med.esr.common.model.person.id.VPIDEntityKey;
import gov.va.med.esr.common.persistent.person.MECPeriodDAO;
import gov.va.med.esr.common.persistent.person.PersonTraitsDAO;
import gov.va.med.esr.service.CommsEmailBulletinService;
import gov.va.med.esr.service.CorrespondenceService;
import gov.va.med.esr.service.DemographicService;
import gov.va.med.esr.service.IRSTransmissionService;
import gov.va.med.esr.service.LookupService;
import gov.va.med.esr.service.PersonIdentityTraits;
import gov.va.med.esr.service.PersonService;
import gov.va.med.esr.service.SystemParameterService;
import gov.va.med.esr.service.trigger.BulletinTrigger;
import gov.va.med.fw.batchprocess.AbstractDataQueryProcess;
import gov.va.med.fw.batchprocess.DataProcessExecutionContext;
import gov.va.med.fw.batchprocess.DataQueryProcessExecutionContext;
import gov.va.med.fw.service.ServiceException;
import gov.va.med.fw.util.StringUtils;

/**
 * Batch process to generate bulk mailing request for IRS form 1095B and ACA letters
 * Reads from pending correspondence queue for type ACA letter and 1095B for the tax year
 * and generates data file to be transferred to NPC for print
 *
 * Create 28 Mar 2016
 * @author DNS   faulkj
 *
 * @version 1.0
 */
public class GenerateACAMailExtract extends AbstractDataQueryProcess {

	private PersonService personService;
	private CorrespondenceService correspondenceService;
	private DemographicService demographicService;
	private LookupService lookupService;
	private MECPeriodDAO mecPeriodDAO = null;
	private PersonTraitsDAO personTraitsDAO = null;
	private IRSTransmissionService irsTransmissionService = null;
	private CommsEmailBulletinService bulletinService = null;
	public final int DEFAULT_JOB_RESULT_UPDATE_INTERVAL = 100;
	private String outputFileLocation;
	public static final String CONTEXT_TASK_COUNT = "taskCount";
	public static final String ERR_DUPE = "Duplicate Mailing Request";
	public static final String ERR_ADDRESS = "Letter Address is Invalid or Missing";
	public static final String ERR_SUPPRESS_DECEASED = "Mail Correspondence Suppressed (Presumed Deceased)";
	public static final String ERR_RETRIEVE = "Failed to Retrieve Record";
	public static final String formnumA = "FORM=742-800A";
	public static final String formnumD = "FORM=742-800D";
	private Writer extractWriter;
	private Writer correctionWriter;
	private boolean onLineTraits = false;
	private String year;
	ACAExportStatistics origStats = null;
	ACAExportStatistics corrStats = null;

	protected void executeProcess(DataProcessExecutionContext context) throws Exception {

		onLineTraits = false;

		String args = (String)context.getExecutionArguments();
		if (!StringUtils.isEmpty(args)) {

			onLineTraits = true;

		}

		//mailings may exist for current year, next year, or up to 4 years back
		//each file must contain only records for a distinct year
		//each file must be named for that distinct year,
		//and each can contain only one type either correction or original, cannot mix
		Calendar cal = GregorianCalendar.getInstance();

		cal.add(Calendar.YEAR, -4);
		int lastfour = cal.get(Calendar.YEAR);
		cal.add(Calendar.YEAR, 1);
		int lastthree = cal.get(Calendar.YEAR);
		cal.add(Calendar.YEAR, 1);
		int lasttwo = cal.get(Calendar.YEAR);
		cal.add(Calendar.YEAR, 1);
		int lastone = cal.get(Calendar.YEAR);
		cal.add(Calendar.YEAR, 1);
		int curYear = cal.get(Calendar.YEAR);
		cal.add(Calendar.YEAR, 1);
		int nextYear = cal.get(Calendar.YEAR);



		String[] years = {String.valueOf(lastfour), String.valueOf(lastthree),
				String.valueOf(lasttwo), String.valueOf(lastone), String.valueOf(curYear), String.valueOf(nextYear)};

		for (int i = 0; i < years.length; i ++) {
			year = years[i];

			this.setParamNames(new String[] {"year"});
			this.setParamValues(new String[] {years[i]});

			origStats = new ACAExportStatistics();
			corrStats = new ACAExportStatistics();

			super.executeProcess(context);
		}
	}

	@SuppressWarnings("rawtypes")
	protected void processData(DataProcessExecutionContext context, List acquiredData) {
		//format used in the mail file as the generation date for each row
		SimpleDateFormat sdf2 = new SimpleDateFormat("MMddyyyy");
		String shortdate = sdf2.format(new Date());

		if (acquiredData == null || acquiredData.size() == 0){
			return;
		}

		for (int i = 0; i < acquiredData.size(); i++) {

			Correspondence form = null;
			Correspondence letter = null;
			boolean isDeceasedType = false;
			boolean isCorrectionType = false;
			Person onFile;

			try {

				if (isInterrupted(context)) {
					this.handleDataProcessCompleted(context);
				}

				BigDecimal personId = (BigDecimal) acquiredData.get(i);
				PersonIdEntityKey personKey = CommonEntityKeyFactory.createPersonIdEntityKey(personId);

				VPIDEntityKey vpid = this.getPersonService().getVPIDByPersonId(personKey);

				//execution param in case we have to use DAO to get traits for bulk mailing
				if (onLineTraits) {
					onFile = this.getPersonService().getPerson(vpid);
				} else {
					onFile = this.getPersonService().getPersonWithoutIdentityTraits(vpid);
				}

				if (onFile == null) {
					setRejectAndError(letter, form, personId.toString(), ERR_RETRIEVE, context);
					origStats.incrementNumberErrIdentity();
					continue;
				}

				//determine type of letter
				//letter is associated only at request time so death info is always most current
				isDeceasedType = determineLetterType(onFile);

				ArrayList<Correspondence> entries = this.getCorrespondenceService().
						getPendingPersonCorrespondence(personId, ComLetterTemplateType.FORM_NUMBER_800);

				//remove dupes
				if (entries != null && entries.size() > 1) {
					//if more than one pending from800 for same person, keep the first one and reject the others as duplicate
					//this should be rare to have any dupes pending, but we have to ensure prevent duplicate entries in the mail file
					//only keep the most recent one and local reject the others
					Collections.sort(entries, new Comparator<Correspondence>() {
						  public int compare(Correspondence o1, Correspondence o2) {
						      if (o1.getStatusDate() == null || o2.getStatusDate() == null)
						        return 0;
						      return o1.getStatusDate().compareTo(o2.getStatusDate());
						  }
						});

					Collections.reverse(entries);

					form = entries.get(0);

					for (int x = 1; x < entries.size(); x++) {
						Correspondence dupe = entries.get(x);
						setRejectAndError(null, dupe, onFile.getPersonEntityKey().getKeyValueAsString(), ERR_DUPE, context);
						origStats.incrementNumberErrDuplicate();
					}

				} else if (entries != null && entries.size() == 1) {
					form = entries.get(0);
				}

				if (entries == null || form == null) {
					setRejectAndError(letter, form, onFile.getPersonEntityKey().getKeyValueAsString(), ERR_RETRIEVE, context);
					origStats.incrementNumberErrOther();
					continue;
				}

				Address mailingAddress = this.getDemographicService().getLetterAddress(onFile);

				form.setMailingAdress(mailingAddress);

				this.getCorrespondenceService().updateACACorrespondence(form, CorrespondenceStatus.SENT_TO_CMS, null);

				//initialize Original or Correction file writer if needed
				isCorrectionType = determineCorrectionType(form);
				initializeWriters(isCorrectionType);

				//insert specific letter type to accompany the 1095 print
				letter = new Correspondence(onFile);
				letter.setEffectiveDate(form.getEffectiveDate());
				letter.setAddressId(form.getAddressId());

				if (isDeceasedType) {
					this.getCorrespondenceService().associateACALetterType(letter, ComLetterTemplateType.FORM_NUMBER_800D);
					if (isCorrectionType) {
						corrStats.incrementNumberForm800D();
					} else {
						origStats.incrementNumberForm800D();
					}
				} else {
					this.getCorrespondenceService().associateACALetterType(letter, ComLetterTemplateType.FORM_NUMBER_800A);
					if (isCorrectionType) {
						corrStats.incrementNumberForm800A();
					} else {
						origStats.incrementNumberForm800A();
					}
				}

				try {
					//validate address
					if (!isAddressValid(mailingAddress)) {
						setRejectAndError(letter, form, onFile.getPersonEntityKey().getKeyValueAsString(), ERR_ADDRESS, context);
						if (isCorrectionType) {
							corrStats.incrementNumberErrAddress();
						} else {
							origStats.incrementNumberErrAddress();
						}
						continue;
					}

					//validate should suppress
					//only if triggered automatically by the job and not by a user
					if (form.getCreatedBy().getName().contains("Job-")) {
						if (this.getCorrespondenceService().shouldSuppressMailCorrespondence(onFile)) {
							setRejectAndError(letter, form, onFile.getPersonEntityKey().getKeyValueAsString(), ERR_SUPPRESS_DECEASED, context);
							if (isCorrectionType) {
								corrStats.incrementNumberErrSuppress();
							} else {
								origStats.incrementNumberErrSuppress();
							}
							continue;
						}
					}

				} catch (Exception ex){
					setRejectAndError(letter, form, onFile.getPersonEntityKey().getKeyValueAsString(), ex.getMessage(), context);
					if (isCorrectionType) {
						corrStats.incrementNumberErrOther();
					} else {
						origStats.incrementNumberErrOther();
					}
					continue;

				}

				try {

					//for corrections, get the requested corrected period from irs transmission data entry
					IrsTransmissionLogDetailEntry submittedEntry = null;
					if (isCorrectionType) {
						submittedEntry = this.getIrsTransmissionService().getMostRecentTransmissionForYear
				    			(onFile.getPersonEntityKey(), year, true);
					}
					//if nothing returned, something went wrong with irs submit or
					//corr entry was inserted in error, no point to continue without a correction period
					if (isCorrectionType && submittedEntry == null) {
						setRejectAndError(letter, form, onFile.getPersonEntityKey().getKeyValueAsString(), ERR_RETRIEVE, context);
						corrStats.incrementNumberErrOther();
						continue;
					}

					//create the full data row
					String dataLine = "";
					if (isDeceasedType) {
						dataLine += formnumD + "^" + shortdate + "^";
					} else {
						dataLine += formnumA + "^" + shortdate + "^";
					}

					//separate because we might have to change where traits are coming from
					dataLine += genTraitsData(onFile);
					//address
					dataLine += genAddressData(onFile, mailingAddress);

					//MEC Period, takes from MEC DAO if original, from IRS transmit if correction
					if (isCorrectionType) {
						String mecData = genMECData(onFile, submittedEntry.getSubmittedData().getMecPeriod());
						if (mecData == null) {
							throw new Exception("Invalid Coverage Period for Tax Year");
						}
						dataLine += mecData;
						//write the row
						synchronized (correctionWriter) {
							correctionWriter.write(dataLine + "\n");
							corrStats.incrementNumberSent();
						}
					} else {
						String mecData =  genMECData(onFile, null);
						if (mecData == null) {
							throw new Exception("Invalid Coverage Period for Tax Year");
						}
						dataLine +=mecData;
						//write the row
						synchronized (extractWriter) {
							extractWriter.write(dataLine + "\n");
							origStats.incrementNumberSent();
						}
					}

					context.getProcessStatistics().incrementNumberOfSuccessfulRecords();

				}  catch (Exception ex) {
					setRejectAndError(letter, form, onFile.getPersonEntityKey().getKeyValueAsString(), ex.getMessage(), context);
					if (isCorrectionType) {
						corrStats.incrementNumberErrOther();
					} else {
						origStats.incrementNumberErrOther();
					}
					continue;
				}

			} catch (Exception ex) {
				setRejectAndError(letter, form, null, ex.getMessage(), context);
				if (isCorrectionType) {
					if (corrStats != null) corrStats.incrementNumberErrOther();
				} else {
					if (origStats != null) origStats.incrementNumberErrOther();
				}
				continue;
			}


			if (i % this.DEFAULT_JOB_RESULT_UPDATE_INTERVAL == 0) {
				this.updateJobResult(context);
			}

 		}

		try {
			if (extractWriter != null) {
				extractWriter.flush();
				extractWriter.close();
			}
			if (correctionWriter != null) {
				correctionWriter.flush();
				correctionWriter.close();
			}
		} catch (Exception ex) {
			logger.error("Unknown Exception in GenerateACAMailExtract:" + ex.getMessage());
			context.getProcessStatistics().incrementNumberOfErrorRecords();
		}


		extractWriter = null;
		correctionWriter = null;

	}

	//builds the section for MEC PERIOD/coverageMonths
	private String genMECData(Person onFile, MECPeriod mp) {
		String mecSegment = "";

		MECPeriod finalone = null;
		if (mp != null) {
			finalone = mp;
		} else {
			finalone = new MECPeriod();
			finalone = this.getIrsTransmissionService().combineAllPeriods((BigDecimal)onFile.getPersonEntityKey().getKeyValue(), Integer.valueOf(year));
		}
		if (finalone.isCoveredAll12Months() == false && finalone.getCoverageMonths().isEmpty()) {
			return null;
		}

		if (finalone.isCoveredAll12Months()) {
			mecSegment += "H01=Y^H02=^H03=^H04=^H05=^H06=^H07=^H08=^H09=^H10=^H11=^H12=^H13=";

		 } else {
			 mecSegment += "H01=^";
			 mecSegment += "H02=" + ( finalone.getCoverageMonths().containsKey("1") && (Boolean) finalone.getCoverageMonths().get("1") ? "Y" : "") + "^";
			 mecSegment += "H03=" + ( finalone.getCoverageMonths().containsKey("2") && (Boolean) finalone.getCoverageMonths().get("2") ? "Y" : "") + "^";
			 mecSegment += "H04=" + ( finalone.getCoverageMonths().containsKey("3") && (Boolean) finalone.getCoverageMonths().get("3") ? "Y" : "") + "^";
			 mecSegment += "H05=" + ( finalone.getCoverageMonths().containsKey("4") && (Boolean) finalone.getCoverageMonths().get("4") ? "Y" : "") + "^";
			 mecSegment += "H06=" + ( finalone.getCoverageMonths().containsKey("5") && (Boolean) finalone.getCoverageMonths().get("5") ? "Y" : "") + "^";
			 mecSegment += "H07=" + ( finalone.getCoverageMonths().containsKey("6") && (Boolean) finalone.getCoverageMonths().get("6") ? "Y" : "") + "^";
			 mecSegment += "H08=" + ( finalone.getCoverageMonths().containsKey("7") && (Boolean) finalone.getCoverageMonths().get("7") ? "Y" : "") + "^";
			 mecSegment += "H09=" + ( finalone.getCoverageMonths().containsKey("8") && (Boolean) finalone.getCoverageMonths().get("8") ? "Y" : "") + "^";
			 mecSegment += "H10=" + ( finalone.getCoverageMonths().containsKey("9") && (Boolean) finalone.getCoverageMonths().get("9") ? "Y" : "") + "^";
			 mecSegment += "H11=" + ( finalone.getCoverageMonths().containsKey("10") && (Boolean) finalone.getCoverageMonths().get("10") ? "Y" : "") + "^";
			 mecSegment += "H12=" + ( finalone.getCoverageMonths().containsKey("11") && (Boolean) finalone.getCoverageMonths().get("11") ? "Y" : "") + "^";
			 mecSegment += "H13=" + ( finalone.getCoverageMonths().containsKey("12") && (Boolean) finalone.getCoverageMonths().get("12") ? "Y" : "");
		}

		return mecSegment;
	}

	//builds address section
	private String genAddressData(Person onFile, Address letterAddress) {
		String eeSegment = "";

		eeSegment += "B01=" + (letterAddress.getLine1() != null ? letterAddress.getLine1() : "") + "^";
		eeSegment += "B02=" + (letterAddress.getLine2() != null ? letterAddress.getLine2() : "") + "^";
		eeSegment += "B03=" + (letterAddress.getLine3() != null ? letterAddress.getLine3() : "") + "^";
		eeSegment += "B04=" + (letterAddress.getCity() != null ? letterAddress.getCity() : "") + "^";
		eeSegment += "B05=" + (letterAddress.getState() != null ? letterAddress.getState() : "") + "^";
		eeSegment += "B06=" + (letterAddress.getCountry() != null ? letterAddress.getCountry() : "USA") + "^";
		eeSegment += "B07=" + (letterAddress.getZipCode() != null ? letterAddress.getZipCode() : "") + (letterAddress.getZipPlus4() != null ? "-" + letterAddress.getZipPlus4() : "") + "^";
		eeSegment += "B08=" + (letterAddress.getPostalCode() != null ? letterAddress.getPostalCode() : "") + "^";
		eeSegment += "B09=" + (letterAddress.getCounty() != null ? letterAddress.getCounty() : "") + "^";
		eeSegment += "B10=" + (letterAddress.getProvince() != null ? letterAddress.getProvince() : "") + "^";

		return eeSegment;
	}

	//put this separate in case we have to fallback to DB lookup
	//builds identity section
	private String genTraitsData(Person onFile) throws Exception {


		String traitsSegment = "";

		if (!onLineTraits) {
			PersonTraits pt = this.getPersonTraitsDAO().getPersonTraitsByVPID(onFile.getVPIDValue());

			if (pt == null) throw new Exception("Failed to Retrieve Person");

			PersonIdentityTraits traits = new PersonIdentityTraits();

			Name name = new Name();
    		name.setFamilyName(pt.getFamilyName());
    		name.setGivenName(pt.getGivenName());
    		name.setMiddleName(pt.getMiddleName());
    		name.setType(lookupService.getNameTypeByCode(NameType.LEGAL_NAME.getName()));
    		traits.addName(name);

    		SSN ssn = new SSN();
    		ssn.setSsnText(pt.getSsn());
    		ssn.setType(lookupService.getSSNTypeByCode(SSNType.CODE_ACTIVE.getName()));
    		traits.setSsn(ssn);

    		BirthRecord birthRecord = new BirthRecord();
    		birthRecord.setBirthDate(new ImpreciseDate(pt.getBirthDate()));
    		traits.setBirthRecord(birthRecord);

    		traits.setGender(lookupService.getGenderByCode(pt.getGender()));

			onFile.setIdentityTraits(traits);
		}

		traitsSegment += "A00=" + (onFile.getIdentityTraits().getLegalName().getPrefix() !=null ? onFile.getIdentityTraits().getLegalName().getPrefix():"") + "^";
		traitsSegment += "A01=" + (onFile.getIdentityTraits().getLegalName().getFamilyName() != null ? onFile.getIdentityTraits().getLegalName().getFamilyName() : "")  + "^";
		traitsSegment += "A02=" + (onFile.getIdentityTraits().getLegalName().getGivenName() != null ? onFile.getIdentityTraits().getLegalName().getGivenName() : "")  + "^";
		traitsSegment += "A03=" + (onFile.getIdentityTraits().getLegalName().getMiddleName() != null ? onFile.getIdentityTraits().getLegalName().getMiddleName() : "") + "^";
		traitsSegment += "A04=" + (onFile.getIdentityTraits().getLegalName().getSuffix() != null ? onFile.getIdentityTraits().getLegalName().getSuffix() : "")  + "^";

		traitsSegment += "A05=" + (onFile.getIdentityTraits().getGender() != null ? onFile.getIdentityTraits().getGender().getCode() : "")  + "^";
		traitsSegment += "A15=" + onFile.getVPIDValue() + "^";
		traitsSegment += "A16=" + onFile.getIdentityTraits().getSsnText() + "^";
		traitsSegment += "N03=" + onFile.getBirthRecord().getBirthDate().getyyyyMMddFormat() + "^";

		return traitsSegment;
	}

	static int getTaskCount(DataProcessExecutionContext context) {
		Integer count = (Integer) context.getContextData().get(CONTEXT_TASK_COUNT);
		return count != null ? count.intValue() : 0;
	}

	//generic exception flow
	//set entries to rejected to avoid reprocessing and hung up send statuses
	private void setRejectAndError(Correspondence letter, Correspondence form, String personId, String errMsg, DataProcessExecutionContext context) {
		try {
			if (letter != null)
			this.getCorrespondenceService().updateACACorrespondence(letter, CorrespondenceStatus.ENROLLMENT_REJECT, errMsg);

			if (form != null)
			this.getCorrespondenceService().updateACACorrespondence(form, CorrespondenceStatus.ENROLLMENT_REJECT, errMsg);

			context.getProcessStatistics().incrementNumberOfErrorRecords();

			logger.error("Failed to generate ACA extract mail entry for person: " + personId + ": " + errMsg);

		} catch (ServiceException ex) {
			logger.error("Failed to update ACA correspondence for person: " + personId, ex);
		}
	}

	//sometimes will be writing both correction and orginals in same run
	//sometimes only one or the other
	//create file at runtime only when come across record of that type for the first time, for that year
	private void initializeWriters(boolean isCorrection) {
		if (isCorrection) {
			if (correctionWriter == null) {
		   		 Date now = new Date();
		   		 SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
		   		 String ts = sdf.format(now);
		   		 String fileName = "MEC_DataExtract_C_" + year + "_V_" + ts + ".txt";
		   		 correctionWriter = getOutputWriter(this.getOutputFileLocation(), fileName);
		   		 corrStats.setFilename(fileName);
		   		 corrStats.setStartTime(now);
			}
		} else {
			if (extractWriter == null) {
		   		 Date now = new Date();
		   		 SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
		   		 String ts = sdf.format(now);
		   		 String fileName = "MEC_DataExtract_O_" + year + "_V_" + ts + ".txt";
		   		 extractWriter = getOutputWriter(this.getOutputFileLocation(), fileName);
		   		 origStats.setFilename(fileName);
		   		 origStats.setStartTime(now);
			}
		}
	}

	private boolean determineLetterType(Person onFile) {
		//if deceased but still covered during some part of the tax year
		//then we have to mail 800D letter type
		//return true if we should send the deceased version of the letter
		//false otherwise
		if (onFile.getDeathRecord() != null && onFile.getDeathRecord().getDeathDate() != null) {
			return true;
		}

		if (onFile.getEnrollmentDetermination().getEnrollmentStatus().getCode().
				equals(EnrollmentStatus.CODE_DECEASED.getName())) {
			return true;
		}

		return false;
	}

	//effective date for originals always set as jan 1st of tax year
	//corrections will always be after this day
	private boolean determineCorrectionType(Correspondence corr) {

		Calendar cal = Calendar.getInstance();
		cal.setTime(corr.getEffectiveDate());

		if (cal.get(Calendar.DAY_OF_YEAR) > 1) {
			return true;
		}

		return false;
	}

	//must have bare minimum line 1 and state or province for NPC CASS to have a chance
	private boolean isAddressValid(Address address) {

		if (address == null) {
			return false;
		}

		if (address.getLine1() == null || address.getLine1().trim().isEmpty()) return false;
		if (address.getState() == null && address.getProvince() == null) return false;
		if (address.getBadAddressReason() != null) return false;

		return true;
	}


	@SuppressWarnings("unchecked")
	protected void handleDataProcessCompleted(DataProcessExecutionContext  context) {

		//******this gets called for each year, each file or set of files for that year, *************
		//this is not needed since we are not throttling task count
		try {
			if(getTaskCount(context) != 0) {
				synchronized(this) {
					boolean stillProcessing = true;
					while(stillProcessing) {
						wait();
						if(getTaskCount(context) == 0)
							stillProcessing = false;
					}
				}
			}
		} catch(InterruptedException e) {
			throwIllegalStateException("GenerateACAMailExtract was interrupted while it was waiting for " +
					"its tasks to complete", e);
		}

		//update stats
		this.updateJobResult(context);

		//close file writers for extract & exception data
	    try {
	    	if (this.extractWriter != null) {
	    		extractWriter.flush();
	    		extractWriter.close(); }
        } catch (IOException e1) {
        	logger.error("Unable to close Data Extract File" + e1.getMessage());
        }


	    try {
	    	if (this.correctionWriter != null) {
	    		correctionWriter.flush();
	    		correctionWriter.close(); }
        } catch (IOException e1) {
        	logger.error("Unable to close Data Extract File" + e1.getMessage());
        }

        extractWriter = null;
        correctionWriter = null;

        //move completed file to dir where perl scripts transfer to NPC
        //or delete the empty it if all records happend to be rejected during processing
        //unlikely except for small batches of corrections
        if (origStats.getFilename() != null) {
        	File file = new File(this.getOutputFileLocation() + origStats.getFilename());

        	if (origStats.getNumberSent() == 0) {
        		file.delete();
        	} else {

        		file.renameTo(new File(this.getOutputFileLocation() + "/ToCMS/" + origStats.getFilename()));
        	}
        	origStats.setEndTime(new Date());
        }

        if (corrStats.getFilename() != null) {
        	File file = new File(this.getOutputFileLocation() + corrStats.getFilename());
        	if (corrStats.getNumberSent() == 0) {
        		file.delete();
        	} else {
        		file.renameTo(new File(this.getOutputFileLocation() + "/ToCMS/" + corrStats.getFilename()));
        	}

        	corrStats.setEndTime(new Date());
        }


		//email bulletin for the print order
        //separate bulletin for each type, since two separate files
        if (origStats.getFilename() != null && origStats.getNumberSent() > 0) {
        	CommsEmailBulletinService emailSrv = this.getBulletinService();
    		Map bulletinData = new HashMap();
    		bulletinData.put("CompletedDateTime", origStats.getEndTime());
    		bulletinData.put("NumberOfFiles", 1);
    		bulletinData.put("NumberLettersSent", origStats.getNumberSent());
    		bulletinData.put("NumberLetterRejected", origStats.getNumberRejected());
    		bulletinData.put("CommunicationsPerFormTxt", "Total Form 741-800A: " + origStats.getNumberForm800A() + "\n" + "Total Form 741-800D: " + origStats.getNumberForm800D());
    		bulletinData.put("RejectPerReasonPerFormTxt", "Address Error: " + origStats.getNumberErrAddress() +
    				"\n" + "Duplicate Mailing: " + origStats.getNumberErrDuplicate() +
    				"\n" + "Identity Traits not Found: " + origStats.getNumberErrIdentity() +
    				"\n" + "Presumed Deceased: " + origStats.getNumberErrSuppress() +
    				"\n" + "Other: " + origStats.getNumberErrOther());
    		bulletinData.put("FileNameRecordCountTxt", origStats.getFilename() + "\t" + origStats.getNumberSent());
    		try {
    			emailSrv.sendEmailBulletin(BulletinTrigger.DataType.MEC_LETTER_PROCESSING, bulletinData, null);
    		} catch (ServiceException ex) {
    			logger.error("sendNotifySubmissionFailed - cannot send email notification" + ex.getMessage());
    		}
        }

        if (corrStats.getFilename() != null && corrStats.getNumberSent() > 0) {
        	CommsEmailBulletinService emailSrv = this.getBulletinService();
    		Map bulletinData = new HashMap();
    		bulletinData.put("CompletedDateTime", corrStats.getEndTime());
    		bulletinData.put("NumberOfFiles", 1);
    		bulletinData.put("NumberLettersSent", corrStats.getNumberSent());
    		bulletinData.put("NumberLetterRejected", corrStats.getNumberRejected());
    		bulletinData.put("CommunicationsPerFormTxt", "Total Form 741-800A: " + corrStats.getNumberForm800A() + "\n" + "Total Form 741-800D: " + corrStats.getNumberForm800D());
    		bulletinData.put("RejectPerReasonPerFormTxt", "Address Error: " + corrStats.getNumberErrAddress() +
    				"\n" + "Duplicate Mailing: " + corrStats.getNumberErrDuplicate() +
    				"\n" + "Identity Traits not Found: " + corrStats.getNumberErrIdentity() +
    				"\n" + "Presumed Deceased: " + corrStats.getNumberErrSuppress() +
    				"\n" + "Other: " + corrStats.getNumberErrOther());
    		bulletinData.put("FileNameRecordCountTxt", corrStats.getFilename() + "\t" + corrStats.getNumberSent());
    		try {
    			emailSrv.sendEmailBulletin(BulletinTrigger.DataType.MEC_LETTER_PROCESSING, bulletinData, null);
    		} catch (ServiceException ex) {
    			logger.error("sendNotifySubmissionFailed - cannot send email notification" + ex.getMessage());
    		}
        }

       super.handleDataProcessCompleted(context);

	}

	private Writer getOutputWriter(String fileLocation, String fileName)
	{
		Writer output = null;
		try {
			File file = new File(fileLocation + fileName);
			output = new BufferedWriter(new FileWriter(file));
		} catch (IOException e) {

		    throw new RuntimeException("GenerateACAMailExtract is unable to generate file to CMS! File= " + fileLocation + fileName+ "Exception: " + e);
		}

		return output;
	}

	protected boolean shouldUpdateJobResult(DataQueryProcessExecutionContext context) {
		return context.getProcessStatistics().isTotalNumberMod(DEFAULT_JOB_RESULT_UPDATE_INTERVAL);
	}

	public PersonService getPersonService() {
		return personService;
	}
	public void setPersonService(PersonService personService) {
		this.personService = personService;
	}

	public CorrespondenceService getCorrespondenceService() {
		return correspondenceService;
	}
	public void setCorrespondenceService(CorrespondenceService correspondenceService) {
		this.correspondenceService = correspondenceService;
	}


	public DemographicService getDemographicService() {
		return demographicService;
	}

	public void setDemographicService(DemographicService demographicService) {
		this.demographicService = demographicService;
	}

	public String getOutputFileLocation() {
		return outputFileLocation;
	}

	public void setOutputFileLocation(String outputFileLocation) {
		this.outputFileLocation = outputFileLocation;
	}

	public LookupService getLookupService() {
		return lookupService;
	}

	public void setLookupService(LookupService lookupService) {
		this.lookupService = lookupService;
	}

	public MECPeriodDAO getMecPeriodDAO() {
		return mecPeriodDAO;
	}

	public void setMecPeriodDAO(MECPeriodDAO mecPeriodDAO) {
		this.mecPeriodDAO = mecPeriodDAO;
	}

	public IRSTransmissionService getIrsTransmissionService() {
		return irsTransmissionService;
	}

	public void setIrsTransmissionService(
			IRSTransmissionService irsTransmissionService) {
		this.irsTransmissionService = irsTransmissionService;
	}

	public PersonTraitsDAO getPersonTraitsDAO() {
		return personTraitsDAO;
	}

	public void setPersonTraitsDAO(PersonTraitsDAO personTraitsDAO) {
		this.personTraitsDAO = personTraitsDAO;
	}

	public CommsEmailBulletinService getBulletinService() {
		return bulletinService;
	}

	public void setBulletinService(CommsEmailBulletinService bulletinService) {
		this.bulletinService = bulletinService;
	}


}
