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.ProductServiceException;
import com.prime.sony.ecommerce.model.Product;
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 ProductBatch {
	private static final Logger log = LoggerFactory.getLogger(ProductBatch.class);

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

	@Autowired
	private MailService mailService;

	@Autowired
	private Connection dataBaseConnection;

	public List<BatchInfo> createBatche(BulkConnection connection, JobInfo jobInfo, List<Product> productList)
			throws IOException, AsyncApiException, ProductServiceException {
		log.info("==>ProductBatch.createBatche()");

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

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

			productFile = File.createTempFile("Product", ".csv");

			try {
				FileOutputStream tmpOut = null; // new FileOutputStream(tmpFile);
				FileOutputStream productOutPutFile = new FileOutputStream(productFile);
				int maxBytesPerBatch = 10000000; // 10 million bytes per batch
				int maxRowsPerBatch = 10000; // 10 thousand rows per batch
				int currentBytes = 0;
				int currentLines = 1;
				String nextLine;
				Product product = null;

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

				List<String> productsSkipped = new ArrayList();
				for (int i = 0; i < productList.size(); i++) {
					product = productList.get(i);

					if (currentBytes + product.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, product);
					tmpOut.write((product.getCSVvalue() + "\n").getBytes());
					productOutPutFile.write((product.getCSVvalue() + "\n").getBytes());
					currentBytes += product.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 ProductBatch.createBatche()", e);
				throw new ProductServiceException(e.getMessage(), e.getCause(), null);
			} finally {
				// tmpFile.delete();
			}
		}
		log.info("<==ProductBatch.createBatche()");
		return batchInfos;
	}

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

		closeJob(connection, job.getId());
		awaitCompletion(connection, job, batchInfoList);
		Map<String, String> resultMap = checkResults(connection, job, batchInfoList, jobStartDate);
		log.info("<==ProductBatch.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 ProductServiceException
	 */
	private Map<String, String> checkResults(BulkConnection connection, JobInfo job, List<BatchInfo> batchInfoList,
			Date jobStartDate) throws AsyncApiException, IOException, ProductServiceException {
		// batchInfoList was populated when batches were created and submitted
		log.info("==>ProductBatch.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("Product 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<Product> productList = readAndWriteFailureRecordsDataIntoCSV(failureRecordIdList,
							rowCSVBatchMap.get(tempFileCount));

					if (productList.size() > 0) {
						String query = "INSERT INTO LOG_PRODUCT (BATCH_PROCESS_ID, JOB_ID, PRODUCT_PRIMARY_KEY, ERROR_DESCRIPTION, LOG_GENERATED_DATE, STATUS) VALUES (?, ?, ?, ?, ?, ?)";

						PreparedStatement ps = dataBaseConnection.prepareStatement(query);
						for (Product productObj : productList) {
							ps.setString(1, b.getId());
							ps.setString(2, job.getId());
							ps.setLong(3, productObj.getProductPrimaryKey());
							ps.setString(4, productObj.getErrorMessage());
							ps.setTimestamp(5, new Timestamp(new Date().getTime()));
							ps.setString(6, "Error");
							ps.addBatch();
						}
						ps.executeBatch();
					}
				}
			} catch (Exception e) {
				log.info("==>ProductBatch.checkResults()", e);
				throw new ProductServiceException(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 ProductServiceException(e.getMessage(), e.getCause(), null);
		}

		log.debug("PRODUCT DATA INSERTED SUCCESS COUNT ::::::::::::::::: " + successCount);
		log.debug("PRODUCT DATA UPSERT SUCCESS COUNT ::::::::::::::::: " + upsertCount.toString());
		log.debug("PRODUCT DATA INSERTED FAILURE COUNT ::::::::::::::::: " + failureCount);
		log.info("<==ProductBatch.checkResults()");
		resultMap.put("JobID", job.getId());
		resultMap.put("BatchIDs", batchId);
		return resultMap;

	}

	/*
	 * private List<Product> readAndWriteFailureRecordsDataIntoCSV(Map<Integer,
	 * String> failureRecordIdList, Map<String, String> resultMap, File
	 * tempFileData) throws ProductServiceException, IOException {
	 * log.info("==>ProductBatch.readAndWriteFailureRecordsDataIntoCSV()");
	 * FileOutputStream failureRecordsFile = null; File failureRecordsTempFile =
	 * null; List<Product> productList = new ArrayList<Product>(); try {
	 * failureRecordsTempFile = File.createTempFile("FailureRecordsFile", ".csv");
	 * failureRecordsFile = new FileOutputStream(failureRecordsTempFile);
	 * 
	 * 
	 * //failureRecordsFile.write((product.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(",");
	 * Product product = new Product();
	 * 
	 * if (failureRecordIdList.containsKey(count)) { //
	 * failureRecordsFile.write((line + "," + failureRecordIdList.get(count) +
	 * "\n").getBytes()); product.setProductPrimaryKey(Long.parseLong(lineData[0]));
	 * product.setErrorMessage(failureRecordIdList.get(count));
	 * productList.add(product); } count = count + 1; }
	 * 
	 * } catch (Exception e) { log.
	 * error("==> Exception at ProductBatch.readAndWriteFailureRecordsDataIntoCSV()"
	 * , e); throw new ProductServiceException(e.getMessage(), e.getCause(), null);
	 * } finally { failureRecordsFile.flush(); productFile.delete(); //
	 * failureRecordsTempFile.delete(); }
	 * log.info("<==ProductBatch.readAndWriteFailureRecordsDataIntoCSV()"); return
	 * productList;
	 * 
	 * }
	 */

	private List<Product> readAndWriteFailureRecordsDataIntoCSV(Map<Integer, String> failureRecordIdList,
			Map<Integer, Product> productbyRowMap) throws ProductServiceException {
		log.info("==>CustomerBatch.readAndWriteFailureRecordsDataIntoCSV()");

		List<Product> productList = new ArrayList<>();
		try {
			Product product = new Product();
			for (Map.Entry<Integer, String> entry : failureRecordIdList.entrySet()) {
				System.out.println("key " + entry.getKey() + "value =" + entry.getValue());

				if (productbyRowMap.containsKey(entry.getKey())) {

					product = productbyRowMap.get(entry.getKey());

					product.setErrorMessage(entry.getValue());
					productList.add(product);
				}

			}

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

	}

	private void createBatch(FileOutputStream tmpOut, File tmpFile, List<BatchInfo> batchInfos,
			BulkConnection connection, JobInfo jobInfo) throws IOException, AsyncApiException, ProductServiceException {
		log.info("==>ProductBatch.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 ProductBatch.createBatch()", e);
			throw new ProductServiceException(e.getMessage(), e.getCause(), null);
		} finally {
			tmpInputStream.close();
		}
	}

}