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.CustomerServiceException;
import com.prime.sony.ecommerce.model.Customer;
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;

import net.snowflake.client.jdbc.internal.joda.time.DateTime;

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

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

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

	public List<BatchInfo> createBatche(BulkConnection connection, JobInfo jobInfo, List<Customer> customerList)
			throws IOException, AsyncApiException, CustomerServiceException {
		log.info("==>CustomerBatch.createBatche()");

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

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

			
			customerFile = File.createTempFile("Customer", ".csv");

			try {
				FileOutputStream tmpOut = null;//new FileOutputStream(tmpFile);
				FileOutputStream customerOutPutFile = new FileOutputStream(customerFile);
				int maxBytesPerBatch = 10000000; // 10 million bytes per batch
				int maxRowsPerBatch = 10000; // 10 thousand rows per batch
				int currentBytes = 0;
				int currentLines = 1;
				String nextLine;
				Customer customer = null;
				
				//tmpOut.write(headerBytes);
				customerOutPutFile.write(headerBytes);
				//currentBytes = headerBytesLength;

				List<String> customersSkipped = new ArrayList();
				for (int i = 0; i < customerList.size(); i++) {
					customer = customerList.get(i);
					
//					if ((currentBytes + customer.getCSVvalue().getBytes().length) > maxBytesPerBatch
//							|| currentLines > maxRowsPerBatch) {
//						currentBytes = 0;
//						currentLines = 0;
//					}
					
					if (currentBytes + customer.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,customer);
					tmpOut.write((customer.getCSVvalue() + "\n").getBytes());
					customerOutPutFile.write((customer.getCSVvalue() + "\n").getBytes());
					currentBytes += customer.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 CustomerBatch.createBatche()", e);
				throw new CustomerServiceException(e.getMessage(), e.getCause(), null);
			} finally {
				// tmpFile.delete();
			}
		}
		log.info("<==CustomerBatch.createBatche()");
		return batchInfos;
	}

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

		closeJob(connection, job.getId());
		awaitCompletion(connection, job, batchInfoList);
		Map<String, String> resultMap = checkResults(connection, job, batchInfoList, jobStartDate);
		log.info("<==CustomerBatch.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) {
			}
			System.out.println("Awaiting results..." + incomplete.size());
			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 CustomerServiceException
	 */
	private Map<String, String> checkResults(BulkConnection connection, JobInfo job, List<BatchInfo> batchInfoList,
			Date jobStartDate)
			throws AsyncApiException, IOException, CustomerServiceException {
		// batchInfoList was populated when batches were created and submitted
		log.info("==>CustomerBatch.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();
			//batchId += batchId == "" ?  b.getId() : ", " + b.getId();

			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);upsertCount
				}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("Customer 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<Customer> customerList = readAndWriteFailureRecordsDataIntoCSV(failureRecordIdList, rowCSVBatchMap.get(tempFileCount));
					
					//File tempFileData = tempFileList.get(tempFileCount);
					//List<Customer> customerList = readAndWriteFailureRecordsDataIntoCSV(failureRecordIdList, resultMap, tempFileData);
					
					if(customerList.size() > 0) {
					String query = "INSERT INTO LOG_CUSTOMER (BATCH_PROCESS_ID, JOB_ID, CUSTOMER_PRIMARY_KEY, ERROR_DESCRIPTION, LOG_GENERATED_DATE, STATUS) VALUES (?, ?, ?, ?, ?, ?)";
					
					PreparedStatement ps = dataBaseConnection.prepareStatement(query);            
					for (Customer customerObj : customerList) {
					    ps.setString(1, b.getId());
					    ps.setString(2, job.getId());
					    ps.setLong(3, customerObj.getCustomer_primary_key());
					    ps.setString(4, customerObj.getErrorMessage());
					    ps.setTimestamp(5,  new Timestamp(new Date().getTime()));
					    ps.setString(6, "Error");
					    ps.addBatch();
					}
					ps.executeBatch();
					}
				}

			} catch (Exception e) {
				log.info("==>CustomerBatch.checkResults()", e);
				throw new CustomerServiceException(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 CustomerServiceException(e.getMessage(), e.getCause(), null);
		}

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

	}

	/*
	 * private List<Customer> readAndWriteFailureRecordsDataIntoCSV(Map<Integer,
	 * String> failureRecordIdList, Map<String, String> resultMap, File
	 * tempFileData) throws CustomerServiceException, IOException {
	 * log.info("==>CustomerBatch.readAndWriteFailureRecordsDataIntoCSV()");
	 * FileOutputStream failureRecordsFile = null; File failureRecordsTempFile =
	 * null; List<Customer> customerList = new ArrayList<Customer>();
	 * for(Map.Entry<Integer,String> entry:failureRecordIdList.entrySet()) {
	 * System.out.println("key " +entry.getKey()+ "value ="+entry.getValue());
	 * 
	 * }
	 * 
	 * try { failureRecordsTempFile = File.createTempFile("FailureRecordsFile",
	 * ".csv"); failureRecordsFile = new FileOutputStream(failureRecordsTempFile);
	 * 
	 * 
	 * //failureRecordsFile.write((customer.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(",");
	 * Customer customer = new Customer();
	 * 
	 * if (failureRecordIdList.containsKey(count)) { failureRecordsFile.write((line
	 * + "," + failureRecordIdList.get(count) + "\n").getBytes());
	 * customer.setCustomer_primary_key(Long.parseLong(lineData[0]));
	 * customer.setErrorMessage(failureRecordIdList.get(count));
	 * customerList.add(customer); } count = count + 1; }
	 * 
	 * } catch (Exception e) { log.
	 * error("==> Exception at CustomerBatch.readAndWriteFailureRecordsDataIntoCSV()"
	 * , e); throw new CustomerServiceException(e.getMessage(), e.getCause(), null);
	 * } finally { failureRecordsFile.flush(); customerFile.delete(); //
	 * failureRecordsTempFile.delete(); }
	 * log.info("<==CustomerBatch.readAndWriteFailureRecordsDataIntoCSV()"); return
	 * customerList;
	 * 
	 * }
	 */
	
	private List<Customer> readAndWriteFailureRecordsDataIntoCSV(Map<Integer, String> failureRecordIdList,
			Map<Integer, Customer> customerbyRowMap) throws CustomerServiceException {
		log.info("==>CustomerBatch.readAndWriteFailureRecordsDataIntoCSV()");

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

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

	}

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

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

}