package com.mycompany;
/*
 * Licensed Materials - Property of HCL Technologies Limited. (c) Copyright HCL Technologies Limited 1996, 2020.
 */

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.HashMap;
import java.util.Map;

import org.apache.nifi.annotation.behavior.SideEffectFree;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.exception.ProcessException;

import groovy.json.JsonBuilder;

import com.hcl.software.data.ingest.processors.AbstractCommerceExecuteSQLProcessor;


@SideEffectFree
@CapabilityDescription("... short description of this custom processor ...")
public class CustomFieldDocumentProcessor extends AbstractCommerceExecuteSQLProcessor {
	private static final JsonBuilder builder = new JsonBuilder();
    @Override
    protected void init(final ProcessorInitializationContext context) {
    	super.init(context);
		getRelationships().add(RELATIONSHIP_SUCCESS);
		getRelationships().add(RELATIONSHIP_FAILURE);
    }	
	// SQL Used in SQL_SELECT_QUERY
	/*
	SELECT
	    OP.CATENTRY_ID,
	    CASE
	        WHEN OP.PRICE <> 0
	        THEN (OP.PRICE - CP.PRICE) / OP.PRICE * 100
	        ELSE NULL
	    END AS PROFIT_MARGIN
	FROM
	    (SELECT O.CATENTRY_ID CATENTRY_ID, P.PRICE PRICE
	      FROM OFFER O
	         INNER JOIN OFFERPRICE P ON (O.OFFER_ID = P.OFFER_ID AND P.CURRENCY = 'USD')
	         INNER JOIN TRADEPOSCN ON (O.TRADEPOSCN_ID = TRADEPOSCN.TRADEPOSCN_ID AND TRADEPOSCN.NAME = 'Extended Sites Catalog Asset Store')
	      WHERE (O.STARTDATE IS NULL OR CURRENT_TIMESTAMP > O.STARTDATE)
	       AND (O.ENDDATE IS NULL OR O.ENDDATE > CURRENT_TIMESTAMP)
	       AND O.PUBLISHED = 1
	    ) OP,
	    (SELECT OFFER.CATENTRY_ID CATENTRY_ID, OFFERPRICE.PRICE PRICE
	         FROM OFFER
	         INNER JOIN OFFERPRICE ON (OFFER.OFFER_ID = OFFERPRICE.OFFER_ID AND OFFERPRICE.CURRENCY = 'USD')
	         INNER JOIN TRADEPOSCN ON (OFFER.TRADEPOSCN_ID = TRADEPOSCN.TRADEPOSCN_ID AND TRADEPOSCN.NAME = 'My Company Cost Price List' )
	    ) CP
	WHERE OP.CATENTRY_ID = CP.CATENTRY_ID AND OP.CATENTRY_ID IN (${data.catentryId}) ${extCatentryAndSQL}     
	 */
	@Override
	public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
		final ComponentLog LOGGER = getLogger();
		LOGGER.debug("Enter");
		
		FlowFile flowFile = getFlowFile(session);
		if (flowFile == null) {
			return;
		}
		Connection conn = null;
		PreparedStatement st = null;
		ResultSet resultSet = null;
		final boolean isNRT = isNRT(flowFile);
		// catalog entry id to profile margin map
		final Map<String, Double> catentryIdProfitMarginsMap = new HashMap<String, Double>();
		try {
		    	conn = dbcpService.getConnection();
		        String selectQuery = context.getProperty(SQL_SELECT_QUERY).evaluateAttributeExpressions(flowFile).getValue();			
		        st = conn.prepareStatement(selectQuery);	    	
				resultSet = executeSQL(context, flowFile, conn, st);
		        while (resultSet.next()) {
		        	String catentryId = resultSet.getString("CATENTRY_ID");
		        	String profitMargin = resultSet.getString("PROFIT_MARGIN");
		        	if (profitMargin != null) {
		        		catentryIdProfitMarginsMap.put(catentryId, Double.valueOf(profitMargin));
		        	}
		        }
		} catch (Throwable e) {
			LOGGER.warn("Unhandled exception encountered when execute SQL: " + e, e);
			handleThrowable(session, flowFile, null, e);
		}
		finally {
			closeDatabase(conn, st, resultSet);
		}

		try {    
	        if (LOGGER.isDebugEnabled()) {
	        	LOGGER.debug("profit margin size: " + catentryIdProfitMarginsMap.size());
	        }
	        if (catentryIdProfitMarginsMap.size() > 0) {
				// Parse update header and document content in flow file
		        String[] items = getFlowFileContentArray(session, flowFile);
		        if (LOGGER.isDebugEnabled()) {
		        	LOGGER.debug("items size: " + items.length);
		        }
				for (int i = 0; i < items.length; i++) {
					
					String item = items[i];
					// Convert Json string to map
					Map<String, Object> itemMap = getDocumentMap(item);
					if (itemMap != null) {
						// Document to update
						Map<String, Object> doc = itemMap;
						if (isNRT) {
							doc = (Map<String, Object>)itemMap.get("doc");
						}
						if (doc != null) {
							// Extract catalog entry id from document 
							String catentryId = null;
							Map<String, Object> id = (Map<String, Object>)doc.get("id");
							if (id != null) {
								catentryId = (String)id.get("catentry");
							}							
							if (catentryId != null) {
								Double priceMargin = catentryIdProfitMarginsMap.get(catentryId);
								if (priceMargin != null) {
									final Map<String, Double> profitMarignMap = new HashMap<String, Double>();
									profitMarignMap.put("x_profitMargin", priceMargin);
									doc.put("x_custom", profitMarignMap);
									// Convert map to Json string and put it back
									builder.call(itemMap);
									items[i] = builder.toString();
								}
							}
						}
					}
				}
				StringBuilder output = new StringBuilder();
				for (String item : items) {
					output.append(item).append("\n");
				}
				flowFile = sendFlowFile(session, flowFile, output.toString());
				transferFlowFile(session, flowFile, RELATIONSHIP_SUCCESS);
	        }
	        else {
	        	transferFlowFile(session, flowFile, RELATIONSHIP_SUCCESS);
	        }

		} catch (Throwable e) {

			LOGGER.warn("Unhandled exception encountered: " + e, e);
			handleThrowable(session, flowFile, null, e);
		}		
		LOGGER.debug("Exit");
	}
}