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

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

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

	public List<BatchInfo> createBatche(BulkConnection connection, JobInfo jobInfo, List<Orders> ordersList)
			throws IOException, AsyncApiException, OrderServiceException {
		log.info("==>OrderBatch.createBatche()");

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

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

			
			orderFile = File.createTempFile("Order", ".csv");

			try {
				FileOutputStream tmpOut = null; //new FileOutputStream(tmpFile);
				FileOutputStream orderOutPutFile = new FileOutputStream(orderFile);
				int maxBytesPerBatch = 10000000; // 10 million bytes per batch
				int maxRowsPerBatch = 10000; // 10 thousand rows per batch
				int currentBytes = 0;
				int currentLines = 1;
				String nextLine;
				Orders orders = null;

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

				List<String> customersSkipped = new ArrayList();
				for (int i = 0; i < ordersList.size(); i++) {
					orders = ordersList.get(i);

//					if ((currentBytes + customer.getCSVvalue().getBytes().length) > maxBytesPerBatch
//							|| currentLines > maxRowsPerBatch) {
//						currentBytes = 0;
//						currentLines = 0;
//					}

					if (currentBytes + orders.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) {
//						currentBytes = headerBytesLength;
//						currentLines = 1;
//						tmpOut.flush();
//					}

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

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

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

					log.debug("Order 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<Orders> ordersList = readAndWriteFailureRecordsDataIntoCSV(failureRecordIdList, rowCSVBatchMap.get(tempFileCount));
					
					if(ordersList.size() > 0) {
					String query = "INSERT INTO LOG_ORDERS (BATCH_PROCESS_ID, JOB_ID, ORDER_PRIMARY_KEY, ERROR_DESCRIPTION, LOG_GENERATED_DATE, STATUS) VALUES (?, ?, ?, ?, ?, ?)";
					
					PreparedStatement ps = dataBaseConnection.prepareStatement(query);            
					for (Orders orderObj : ordersList) {
					    ps.setString(1, b.getId());
					    ps.setString(2, job.getId());
					    ps.setLong(3, orderObj.getOrder_primary_key());
					    ps.setString(4, orderObj.getErrorMessage());
					    ps.setTimestamp(5,  new Timestamp(new Date().getTime()));
					    ps.setString(6, "Error");
					    ps.addBatch();
					}
					ps.executeBatch();
					}
				}


			} catch (Exception e) {
				log.info("==>OrdersBatch.checkResults()", e);
				throw new OrderServiceException(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("==>OrderBatch.checkResults()", e);
			throw new OrderServiceException(e.getMessage(), e.getCause(), null);
		}
		log.debug("order DATA INSERTED SUCCESS COUNT ::::::::::::::::: " + successCount);
		log.debug("order DATA UPSERT SUCCESS COUNT ::::::::::::::::: " + upsertCount.toString());
		log.debug("order DATA INSERTED FAILURE COUNT ::::::::::::::::: " + failureCount);
		log.info("<==OrdersBatch.checkResults()");
		resultMap.put("JobID", job.getId());
		resultMap.put("BatchIDs", batchId);
		return resultMap;

	}

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

		List<Orders> ordersList = new ArrayList<>();
		try {
			Orders order = new Orders();
			//List<Customer> customerList = new ArrayList<Customer>();
			for(Map.Entry<Integer,String> entry:failureRecordIdList.entrySet()) {
				System.out.println("key " +entry.getKey()+ 
						"value ="+entry.getValue());
				
				if(orderbyRowMap.containsKey(entry.getKey())) {
					
					order = orderbyRowMap.get(entry.getKey());
					
					order.setErrorMessage(entry.getValue());
					ordersList.add(order);
				}
				
			}

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

	}

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

}