package com.prime.sony.ecommerce.helper;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.FileReader;
import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.prime.sony.ecommerce.exceptions.StoreServiceException;
import com.prime.sony.ecommerce.model.Store;
import com.prime.sony.ecommerce.util.MailService;
import com.sforce.async.AsyncApiException;
import com.sforce.async.BatchInfo;
import com.sforce.async.BatchStateEnum;
import com.sforce.async.BulkConnection;
import com.sforce.async.CSVReader;
import com.sforce.async.JobInfo;
import com.sforce.async.JobStateEnum;
import com.sforce.ws.ConnectionException;

@Component
public class StoreBatch {
	private static final Logger log = LoggerFactory.getLogger(StoreBatch.class);

	private static File tmpFile;
	private static File storeFile;
	private static List<File> tempFileList = new ArrayList<File>();
	private static Map<Integer,Map<Integer,Store>> rowCSVBatchMap = new HashMap<>();

	@Autowired
	private MailService mailService;
	
	@Autowired
	private Connection dataBaseConnection;

	public List<BatchInfo> createBatche(BulkConnection connection, JobInfo jobInfo, List<Store> storeList)
			throws IOException, AsyncApiException, StoreServiceException {
		log.info("==>StoreBatch.createBatche()");

		List<BatchInfo> batchInfos = new ArrayList<BatchInfo>();
		int headerBytesLength = 0;
		byte[] headerBytes = null;
		Map<Integer,Store> rowCSVMap = new HashMap<>();
		Integer batchCount = 0;
		rowCSVBatchMap = new HashMap<>();

		if (storeList.size() > 0) {
			headerBytesLength = storeList.get(0).getHeader().split(",").length;
			headerBytes = (storeList.get(0).getHeader() + "\n").getBytes("UTF-8");

			
			storeFile = File.createTempFile("Store", ".csv");

			try {
				FileOutputStream tmpOut = null; //new FileOutputStream(tmpFile);
				FileOutputStream storeOutPutFile = new FileOutputStream(storeFile);
				int maxBytesPerBatch = 10000000; // 10 million bytes per batch
				int maxRowsPerBatch = 10000; // 10 thousand rows per batch
				int currentBytes = 0;
				int currentLines = 1;
				String nextLine;
				Store store = null;

				//tmpOut.write(headerBytes);
				storeOutPutFile.write(headerBytes);
				//currentBytes = headerBytesLength;

				List<String> storesSkipped = new ArrayList();
				for (int i = 0; i < storeList.size(); i++) {
					store = storeList.get(i);

					if (currentBytes + store.getCSVvalue().getBytes().length > maxBytesPerBatch
							|| currentLines > maxRowsPerBatch) {
						rowCSVBatchMap.put(batchCount, rowCSVMap);
						createBatch(tmpOut, tmpFile, batchInfos, connection, jobInfo);
						currentBytes = 0;
						currentLines = 0;
						batchCount++;
						 rowCSVMap = new HashMap<>();
					}

					if (currentBytes == 0) {
						rowCSVMap= new HashMap<>();
						tmpFile = File.createTempFile("bulkAPIInsert", ".csv");
						tmpOut = new FileOutputStream(tmpFile);
						tmpOut.write(headerBytes);
						currentBytes = headerBytesLength;
						currentLines = 1;
					}

					rowCSVMap.put(currentLines,store);
					tmpOut.write((store.getCSVvalue() + "\n").getBytes());
					storeOutPutFile.write((store.getCSVvalue() + "\n").getBytes());
					currentBytes += store.getCSVvalue().getBytes().length;
					currentLines++;
				}
				if (currentLines > 1) {
					// tmpOut.flush();
					rowCSVBatchMap.put(batchCount, rowCSVMap);
					createBatch(tmpOut, tmpFile, batchInfos, connection, jobInfo);
				}

			} catch (Exception e) {
				log.error("==>Exception at StoreBatch.createBatche()", e);
				throw new StoreServiceException(e.getMessage(), e.getCause(), null);
			} finally {
				//tmpFile.delete();
			}
		}
		log.info("<==StoreBatch.createBatche()");
		return batchInfos;
	}

	/**
	 * Creates a Bulk API job and uploads batches for a CSV file.
	 * 
	 * @throws StoreServiceException
	 */
	public Map<String, String> bulkImport(BulkConnection connection, JobInfo job, List<BatchInfo> batchInfoList,
			Date jobStartDate)
			throws AsyncApiException, ConnectionException, IOException, StoreServiceException {
		log.info("==>StoreBatch.bulkImport()");

		closeJob(connection, job.getId());
		awaitCompletion(connection, job, batchInfoList);
		Map<String, String> resultMap = checkResults(connection, job, batchInfoList, jobStartDate);
		log.info("<==StoreBatch.bulkImport()");
		return resultMap;
	}

	private void closeJob(BulkConnection connection, String jobId) throws AsyncApiException {
		JobInfo job = new JobInfo();
		job.setId(jobId);
		job.setState(JobStateEnum.Closed);
		connection.updateJob(job);
	}

	/**
	 * Wait for a job to complete by polling the Bulk API.
	 * 
	 * @param connection
	 *            BulkConnection used to check results.
	 * @param job
	 *            The job awaiting completion.
	 * @param batchInfoList
	 *            List of batches for this job.
	 * @throws AsyncApiException
	 */
	private void awaitCompletion(BulkConnection connection, JobInfo job, List<BatchInfo> batchInfoList)
			throws AsyncApiException {
		long sleepTime = 0L;
		Set<String> incomplete = new HashSet<String>();
		for (BatchInfo bi : batchInfoList) {
			incomplete.add(bi.getId());
		}
		while (!incomplete.isEmpty()) {
			try {
				Thread.sleep(sleepTime);
			} catch (InterruptedException e) {
			}
			sleepTime = 10000L;
			BatchInfo[] statusList = connection.getBatchInfoList(job.getId()).getBatchInfo();
			for (BatchInfo b : statusList) {
				if (b.getState() == BatchStateEnum.Completed || b.getState() == BatchStateEnum.Failed) {
					if (incomplete.remove(b.getId())) {
						System.out.println("BATCH STATUS:\n" + b);
					}
				}
			}
		}
	}

	/**
	 * Gets the results of the operation and checks for errors.
	 * 
	 * @throws StoreServiceException
	 */
	private Map<String, String> checkResults(BulkConnection connection, JobInfo job, List<BatchInfo> batchInfoList,
			Date jobStartDate)
			throws AsyncApiException, IOException, StoreServiceException {
		// batchInfoList was populated when batches were created and submitted
		log.info("==>StoreBatch.checkResults()");
		Integer successCount = 0;
		Integer upsertCount = 0;
		Integer failureCount = 0;
		Map<String, String> resultMap = new HashMap<String, String>();
		
		// Map<String,List<Integer>> failureResultsMap = new
		// HashMap<String,List<Integer>>();
		int tempFileCount = 0;
		String batchId = "";
		for (BatchInfo b : batchInfoList) {
			Map<Integer, String> failureRecordIdList = new HashMap<Integer, String>();
			CSVReader rdr = new CSVReader(connection.getBatchResultStream(job.getId(), b.getId()));
			List<String> resultHeader = rdr.nextRecord();
			int resultCols = resultHeader.size();
			
			List<String> row;
			Integer failureIdCount = 0;
			int failureid = 1;
			while ((row = rdr.nextRecord()) != null) {
				
				Map<String, String> resultInfo = new HashMap<String, String>();
				for (int i = 0; i < resultCols; i++) {
					resultInfo.put(resultHeader.get(i), row.get(i));
				}
				boolean success = Boolean.valueOf(resultInfo.get("Success"));
				boolean created = Boolean.valueOf(resultInfo.get("Created"));
				String id = resultInfo.get("Id");
				String error = resultInfo.get("Error");
				if (success && created) {
					successCount = successCount + 1;
					// System.out.println("Created row with id " + id);
				}else if (success && !created) {
					upsertCount = upsertCount + 1;
					// System.out.println("Created row with id " + id);upsertCount
				} else if (!success) {
					failureCount = failureCount + 1;
					failureIdCount = failureCount;
					failureRecordIdList.put(failureid, error);

					System.out.println("Failed with error: " + error);
					log.debug("Store Data Failed with ERROR: " + error);
				}
				failureid++;
			}
		
		try {
			resultMap.put("SuccessCount", successCount.toString());
			resultMap.put("UpsertCount", upsertCount.toString());
			resultMap.put("FailureCount", failureCount.toString());
			batchId += batchId == "" ?  b.getId() : ", " + b.getId();
			if (failureIdCount > 0 && rowCSVBatchMap != null && rowCSVBatchMap.get(tempFileCount) != null) {
				
				List<Store> storeList = readAndWriteFailureRecordsDataIntoCSV(failureRecordIdList, rowCSVBatchMap.get(tempFileCount));
				
				if(storeList.size() > 0) {
				String query = "INSERT INTO LOG_STORE (BATCH_PROCESS_ID, JOB_ID, STORE_PRIMARY_KEY, ERROR_DESCRIPTION, LOG_GENERATED_DATE, STATUS) VALUES (?, ?, ?, ?, ?, ?)";
				
				PreparedStatement ps = dataBaseConnection.prepareStatement(query);            
				for (Store storeObj : storeList) {
				    ps.setString(1, b.getId());
				    ps.setString(2, job.getId());
				    ps.setString(3, storeObj.getStoreId());
				    ps.setString(4, storeObj.getErrorMessage());
				    ps.setTimestamp(5,  new Timestamp(new Date().getTime()));
				    ps.setString(6, "Error");
				    ps.addBatch();
				}
				ps.executeBatch();
				}
			}

		} catch (Exception e) {
			log.info("==>StoreBatch.checkResults()", e);
			throw new StoreServiceException(e.getMessage(), e.getCause(), null);
		}
		tempFileCount++;
		}
		for(File f : tempFileList) {
			f.delete();
		}
		
		tempFileList = new ArrayList<File>();
		
		try {
			if(Integer.parseInt(resultMap.get("FailureCount")) > 0) {
				mailService.mailsending(resultMap, job.getId(), batchId, jobStartDate);		
			}		
		} catch (Exception e) {
			log.info("==>CustomerBatch.checkResults()", e);
			throw new StoreServiceException(e.getMessage(), e.getCause(), null);
		}
		log.debug("Store DATA INSERTED SUCCESS COUNT ::::::::::::::::: " + successCount);
		log.debug("Store DATA UPSERT SUCCESS COUNT ::::::::::::::::: " + upsertCount.toString());
		log.debug("Store DATA INSERTED FAILURE COUNT ::::::::::::::::: " + failureCount);
		log.info("<==StoreBatch.checkResults()");
		resultMap.put("JobID", job.getId());
		resultMap.put("BatchIDs", batchId);
		return resultMap;

	}

	/*
	 * private List<Store> readAndWriteFailureRecordsDataIntoCSV(Map<Integer,
	 * String> failureRecordIdList, Map<String, String> resultMap, File
	 * tempFileData) throws StoreServiceException, IOException {
	 * log.info("==>StoreBatch.readAndWriteFailureRecordsDataIntoCSV()");
	 * FileOutputStream failureRecordsFile = null; File failureRecordsTempFile =
	 * null; List<Store> storeList = new ArrayList<Store>(); try {
	 * failureRecordsTempFile = File.createTempFile("FailureRecordsFile", ".csv");
	 * failureRecordsFile = new FileOutputStream(failureRecordsTempFile);
	 * 
	 * 
	 * // failureRecordsFile.write((store.getHeader() + ",Error" +
	 * "\n").getBytes()); BufferedReader br = new BufferedReader(new
	 * FileReader(tempFileData)); br.readLine(); String line = ""; int count = 1;
	 * while ((line = br.readLine()) != null) { String[] lineData = line.split(",");
	 * Store store = new Store();
	 * 
	 * if (failureRecordIdList.containsKey(count)) { //
	 * failureRecordsFile.write((line + "," + failureRecordIdList.get(count) +
	 * "\n").getBytes()); store.setStoreId(lineData[0].replaceAll("\"", ""));
	 * store.setErrorMessage(failureRecordIdList.get(count)); storeList.add(store);
	 * } count = count + 1; }
	 * 
	 * } catch (Exception e) { log.
	 * error("==> Exception at StoreBatch.readAndWriteFailureRecordsDataIntoCSV()",
	 * e); throw new StoreServiceException(e.getMessage(), e.getCause(), null); }
	 * finally { failureRecordsFile.flush(); storeFile.delete(); //
	 * failureRecordsTempFile.delete(); }
	 * log.info("<==StoreBatch.readAndWriteFailureRecordsDataIntoCSV()"); return
	 * storeList;
	 * 
	 * }
	 */
	
	private List<Store> readAndWriteFailureRecordsDataIntoCSV(Map<Integer, String> failureRecordIdList,
			Map<Integer, Store> storebyRowMap) throws StoreServiceException {
		log.info("==>StoreBatch.readAndWriteFailureRecordsDataIntoCSV()");

		List<Store> storeList = new ArrayList<>();
		try {
			Store store = new Store();
			for(Map.Entry<Integer,String> entry:failureRecordIdList.entrySet()) {
				System.out.println("key " +entry.getKey()+ 
						"value ="+entry.getValue());
				
				if(storebyRowMap.containsKey(entry.getKey())) {
					
					store = storebyRowMap.get(entry.getKey());
					
					store.setErrorMessage(entry.getValue());
					storeList.add(store);
				}
				
			}

		} catch (Exception e) {
			log.error("==> Exception at StoreBatch.readAndWriteFailureRecordsDataIntoCSV()", e);
			throw new StoreServiceException(e.getMessage(), e.getCause(), null);
		} 
		log.info("<==StoreBatch.readAndWriteFailureRecordsDataIntoCSV()");
		return storeList;

	}

	private void createBatch(FileOutputStream tmpOut, File tmpFile, List<BatchInfo> batchInfos,
			BulkConnection connection, JobInfo jobInfo) throws IOException, AsyncApiException, StoreServiceException {
		log.info("==>StoreBatch.createBatch()");
		tmpOut.flush();
		tmpOut.close();
		FileInputStream tmpInputStream = new FileInputStream(tmpFile);
		try {
			tempFileList.add(tmpFile);
			BatchInfo batchInfo = connection.createBatchFromStream(jobInfo, tmpInputStream);
			batchInfos.add(batchInfo);

		} catch (Exception e) {
			log.error("==>Exception at StoreBatch.createBatch()", e);
			throw new StoreServiceException(e.getMessage(), e.getCause(), null);
		} finally {
			tmpInputStream.close();
		}
	}

}