package kinesisFHM.updateConsumer;

import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.nio.ByteBuffer;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Vector;

import kinesisFHM.commonTypes.DependencyGraph;
import kinesisFHM.commonTypes.DependencyRelation;
import kinesisFHM.commonTypes.GraphUpdate;
import kinesisFHM.commonTypes.WfRType;
import kinesisFHM.utilities.CloudWatchReporter;
import kinesisFHM.utilities.HashTable2D;

import org.apache.commons.codec.binary.Base64;

import com.amazonaws.services.cloudwatch.AmazonCloudWatchClient;
import com.amazonaws.services.cloudwatch.model.Dimension;
import com.amazonaws.services.cloudwatch.model.MetricDatum;
import com.amazonaws.services.cloudwatch.model.PutMetricDataRequest;
import com.amazonaws.services.cloudwatch.model.StandardUnit;
import com.amazonaws.services.kinesis.AmazonKinesisClient;
import com.amazonaws.services.kinesis.model.ExpiredIteratorException;
import com.amazonaws.services.kinesis.model.GetRecordsRequest;
import com.amazonaws.services.kinesis.model.GetRecordsResult;
import com.amazonaws.services.kinesis.model.GetShardIteratorRequest;
import com.amazonaws.services.kinesis.model.GetShardIteratorResult;
import com.amazonaws.services.kinesis.model.ProvisionedThroughputExceededException;
import com.amazonaws.services.kinesis.model.PutRecordsRequest;
import com.amazonaws.services.kinesis.model.PutRecordsRequestEntry;
import com.amazonaws.services.kinesis.model.PutRecordsResult;
import com.amazonaws.services.kinesis.model.Record;
import com.amazonaws.services.kinesis.model.Shard;

public class UpdateConsumerThread extends Thread implements Runnable {

	private String inputStreamName;
	private String outputStreamName;
	private AmazonKinesisClient amazonKinesisClient;
	private CloudWatchReporter inputReporter;
	private CloudWatchReporter outputReporter;
	private Shard inputShard;

	private String fname;
	private Properties props;

	private static float sigma_a;
	private static float sigma_l1l;
	private static float sigma_l2l;
	private static float sigma_ld;
	private static float rel2best;
	
	protected static float defaultThreshold;

	// maximum supported by AWS is 500
	private static final int maxOutputBatchSize = 500;
	// maximum supported by AWS is 1000
	private static final int maxInputBatchSize = 10000;

	private long threadsleep;
	private int inputBatchSize;
	
	private String outputString;
	
	private boolean throttled;

	// these are the dependency tables
	private HashTable2D<String> dep_table = new HashTable2D<String>();
	private HashTable2D<String> l2_table = new HashTable2D<String>();

	// the (partial) dependency graph and the update list to pass on
	private DependencyGraph DG = new DependencyGraph();
	private List<GraphUpdate> updateList = new Vector<GraphUpdate>();
	
	// the updateConsumer object that holds totalCounts
	private UpdateConsumer updateConsumer;

	public UpdateConsumerThread(	
			String inputStreamName, 
			Shard inputShard, 
			AmazonKinesisClient amazonKinesisClient, 
			CloudWatchReporter inputReporter,
			CloudWatchReporter outputReporter,
			String outputStreamName,
			UpdateConsumer updateConsumer,
			long threadsleep,
			int inputBatchSize) {
		
		this.updateConsumer = updateConsumer;
		this.inputStreamName = inputStreamName;
		this.outputStreamName = outputStreamName;
		this.amazonKinesisClient = amazonKinesisClient;
		this.inputReporter = inputReporter;
		this.outputReporter = outputReporter;
		this.inputShard = inputShard;
		this.fname = inputStreamName + "." + inputShard.getShardId();
		this.threadsleep = threadsleep;
		this.inputBatchSize = Math.min(maxInputBatchSize, inputBatchSize);
		this.throttled = false;
		
		File f = new File(fname+".persistedCounts");
		ObjectInputStream ois;
		try {
			ois = new ObjectInputStream(new FileInputStream(f));
			dep_table = (HashTable2D<String>) ois.readObject();
			l2_table = (HashTable2D<String>) ois.readObject();
			DG = (DependencyGraph) ois.readObject();
		} catch (FileNotFoundException e) {
			System.err.println(this + " " + e.getMessage());
		} catch (IOException e) {
			System.err.println(this + " " + e.getMessage());
		} catch (ClassNotFoundException e) {
			System.err.println(this + " " + e.getMessage());
		}
	}
	
	private void adjustThresholds(long t) {
		synchronized (updateConsumer) {
			sigma_ld = sigma_l2l = sigma_l1l = sigma_a = (float) Math.min(1, Math.max(defaultThreshold, Math.log(t) / (Math.log(t) + 1)));
			rel2best = 1 - sigma_a;
			System.err.println(this.getName() + " Total Record Count: " + updateConsumer.totalRecordCount + " == Thresholds adjusted to : " + sigma_ld);
		}
	}


	private DependencyRelation decodeRecord(Record record) {
		byte[] data = Base64.decodeBase64(record.getData().array());
		ByteArrayInputStream bais = new ByteArrayInputStream(data);
		ObjectInputStream ois;
		DependencyRelation rel = null;
		try {
			ois = new ObjectInputStream( bais );
			rel = (DependencyRelation) ois.readObject();
		} catch (Exception e) {
			saveProps();
			System.err.println(this + " " + e.getMessage());
		}
		return rel;
	}

	private void processRecord(Record record) {
		// Decode the record
		DependencyRelation rel = decodeRecord(record);
		
		if (rel != null) {
			// Do a sanity check and don't process illegal ones
			if (rel.getDep() > 1 || rel.getDep() < 0) {
//				System.err.println(this.getName() + " Received illegal dependency relation: " + rel);
				return;
			}

			WfRType type = rel.getType();

			switch (type) {
			case DIRECT : {
				process_direct(rel);
				break;
			}
			case LOOPTWO : {
				process_l2(rel);
				break;
			}
			case SUCCESSOR : {				
				process_ld(rel);
				break;
			}
			case COUNT:
				break;
			default:
//				System.err.println(this.getName() + " Received illegal dependency relation");
				break;
			}
		} else {
//			System.err.println(this.getName() + " Received invalid record");
		}
	}

	private void process_ld(DependencyRelation rel) {
		String ev1 = rel.getEvent1();
		String ev2 = rel.getEvent2();
		Float d = rel.getDep();

		if (d > sigma_ld) {
			if (DG.add(ev1, ev2, WfRType.SUCCESSOR)) updateList.add(new GraphUpdate(ev1, ev2, GraphUpdate.GraphUpdateType.ADD));
		} else {
			if (DG.remove(ev1, ev2, WfRType.SUCCESSOR)) updateList.add(new GraphUpdate(ev1, ev2, GraphUpdate.GraphUpdateType.REMOVE));
		}
	}

	private void process_l2(DependencyRelation rel) {
		String ev1 = rel.getEvent1();
		String ev2 = rel.getEvent2();
		Float d = rel.getDep();

		Float oldL2 = l2_table.get(ev1, ev2);

		if ( (d > oldL2) && (d >= sigma_l2l) && (oldL2 < sigma_l2l)) {
			// gotten bigger and gotten bigger than threshold
			// add to C2, step 3, if conditions are met
			if (DG.add(ev1, ev2, WfRType.LOOPTWO)) updateList.add(new GraphUpdate(ev1, ev2, GraphUpdate.GraphUpdateType.ADD));
			if (DG.add(ev2, ev1, WfRType.LOOPTWO)) updateList.add(new GraphUpdate(ev2, ev1, GraphUpdate.GraphUpdateType.ADD));
		}
		// note the l2loop
		l2_table.set(ev1, ev2, d);
	}

	private void process_direct(DependencyRelation rel) {
		String ev1 = rel.getEvent1();
		String ev2 = rel.getEvent2();
		Float d = rel.getDep();

		Float oldDep = dep_table.set(ev1, ev2, d);

		if (ev1.equals(ev2)) {
			if ( (oldDep < sigma_l1l) && (d >= sigma_l1l) ) {
				// add to C1, step 2
				if (DG.add(ev1, ev2, WfRType.DIRECT)) updateList.add(new GraphUpdate(ev1, ev2, GraphUpdate.GraphUpdateType.ADD));
			}
		} else {
			// Cout, step 4
			Map.Entry<String, Float> Cout = dep_table.rowMaxButNot(ev1, ev2);
			// Cout, the strongest follower of ev1
			if (Cout != null) {
				if (d > Cout.getValue()) {
					// ev2 is the new strongest follower of ev1
					// remove the previous relation from DG
					if (DG.remove(ev1, Cout.getKey(), WfRType.DIRECT)) updateList.add(new GraphUpdate(ev1, Cout.getKey(), GraphUpdate.GraphUpdateType.REMOVE));
					// add the new one
					if (DG.add(ev1, ev2, WfRType.DIRECT)) updateList.add(new GraphUpdate(ev1, ev2, GraphUpdate.GraphUpdateType.ADD));
				}
			} else {
				// ev2 is the only follower of ev1
				if (DG.add(ev1, ev2, WfRType.DIRECT)) updateList.add(new GraphUpdate(ev1, ev2, GraphUpdate.GraphUpdateType.ADD));
			}
			// Cin, step 5
			Map.Entry<String, Float> Cin = dep_table.colMaxButNot(ev2, ev1);
			// Cin, the strongest cause of ev2
			if (Cin != null) {
				if (d > Cin.getValue()) {
					// ev1 is the new strongest cause of ev2
					// remove the previous relation from DG
					if (DG.remove(Cin.getKey(), ev2, WfRType.DIRECT)) updateList.add(new GraphUpdate(Cin.getKey(), ev2, GraphUpdate.GraphUpdateType.REMOVE));
					// add the new one
					if (DG.add(ev1, ev2, WfRType.DIRECT)) updateList.add(new GraphUpdate(ev1, ev2, GraphUpdate.GraphUpdateType.ADD));
				}
			} else {
				// ev1 is the only cause of ev2
				if (DG.add(ev1, ev2, WfRType.DIRECT)) updateList.add(new GraphUpdate(ev1, ev2, GraphUpdate.GraphUpdateType.ADD));
			}

			// Cout_prime, step 6, 7
			for (String v : DG.getTargets(ev1).toArray(new String[]{}) ) {
				if ( DG.contains(v, ev1) && !v.equals(ev2) ) {
					// ev1 is part of an L2L with v, so check whether v has followers that are stronger than ev2
					Map.Entry<String, Float> f = dep_table.rowMaxButNot(v, ev1);
					if (f != null) {
						if (f.getValue() - d > rel2best) {
							// if there is a follower of v that is not back to ev1 and is stronger than ev1->ev2 by at least rel2best then we
							// don't need the loop exit from ev1->ev2, we'll take the loop exit from v instead.
							if (DG.remove(ev1, ev2, WfRType.DIRECT)) updateList.add(new GraphUpdate(ev1, ev2, GraphUpdate.GraphUpdateType.REMOVE));
						}
					}
				}
			}
			// Cin_prime, step 8, 9
			for (String v : DG.getCauses(ev2).toArray(new String[]{}) ) {
				if ( DG.contains(ev2, v) && !v.equals(ev1) ) {
					// ev2 is part of an L2L with v, so check whether v has causes that are stronger than ev1
					Map.Entry<String, Float> f = dep_table.colMaxButNot(v, ev2);
					if (f != null) {
						if (f.getValue() - d > rel2best) {
							// if there is a cause of v that is not from ev1 and is stronger than ev1->ev2 by at least rel2best then we
							// don't need the loop entry from ev1->ev2, we'll take the loop entry from v instead.
							if (DG.remove(ev1, ev2, WfRType.DIRECT)) updateList.add(new GraphUpdate(ev1, ev2, GraphUpdate.GraphUpdateType.REMOVE));
						}
					}
				}
			}	
			// Cout_primeprime, step 10
			if (d > sigma_a) {
				if (DG.add(ev1, ev2, WfRType.DIRECT)) updateList.add(new GraphUpdate(ev1, ev2, GraphUpdate.GraphUpdateType.ADD));
			}
			for (String v : DG.getTargets(ev1).toArray(new String[]{}) ) {
				if (!v.equals(ev2)) {
					if (dep_table.get(ev1, v) - d < rel2best) { 
						if (DG.add(ev1, ev2, WfRType.DIRECT)) updateList.add(new GraphUpdate(ev1, ev2, GraphUpdate.GraphUpdateType.ADD));
					}
				}
			}
			// Cin_primeprime is not processed as part of this relation, but only when the inverse is processed, step 11
		}
	}

	private void emitGraphUpdatesToStream() {
		int writeCount = updateList.size();
		outputString = new String(this.getName() + " emitted " + writeCount + " ");
		while (!updateList.isEmpty()) {
//			System.err.println(this.getName() + " Output Queue size = " + updateList.size());
			PutRecordsRequest putRecordsRequest  = new PutRecordsRequest();
			putRecordsRequest.setStreamName( outputStreamName );
			List <PutRecordsRequestEntry> putRecordsRequestEntryList  = new ArrayList<>(); 

			for (int i=0; i<Math.min(updateList.size(), maxOutputBatchSize); i++) {
				GraphUpdate upd = updateList.get(i);
			    PutRecordsRequestEntry putRecordsRequestEntry  = new PutRecordsRequestEntry();		    
				putRecordsRequestEntry.setData(ByteBuffer.wrap(upd.getBase64()));
				putRecordsRequestEntry.setPartitionKey(upd.getEvent1() + ":" + upd.getEvent2());
				putRecordsRequestEntryList.add(putRecordsRequestEntry);
//				System.err.println(this.getName() + " Added GraphUpdate " + upd + " to " + outputStreamName);
			}
			putRecordsRequest.setRecords(putRecordsRequestEntryList);
			PutRecordsResult putRecordsResult  = amazonKinesisClient.putRecords(putRecordsRequest);
			outputString += ".";
			for(int j=putRecordsResult.getRecords().size()-1; j >= 0 ; j--) {
				if (putRecordsResult.getRecords().get(j).getErrorCode() == null)
					updateList.remove(j);
				else {
					if (putRecordsResult.getRecords().get(j).getErrorCode().equals("ProvisionedThroughputExceededException")) {
						// slowing down by lengthening our sleep period
//						threadsleep++;
						throttled = true;
						System.out.println("Backing off");
					} else {
						System.out.println(this.getName() + " ERROR : " + putRecordsResult.getRecords().get(j).getErrorMessage());
					}
				}
			}
			if (throttled) {
				System.out.println("Trace Output Throttled - Sleeping");
				try { Thread.sleep(threadsleep); } catch (InterruptedException e) {}
			}
		}
		// log the puts to cloudwatch
		outputReporter.add(writeCount);

		if (throttled) 
			threadsleep*=1.05;
		updateList.clear();
	}

	private void loadProps() {
		props = new Properties();
		try {
			props.load(new FileReader(fname + ".properties"));
		} catch (FileNotFoundException e) {
		} catch (IOException e) {
			System.err.println(this + " " + e.getMessage());
		}	
	}

	private void saveProps() {
		try {
			props.store(new FileWriter(fname + ".properties"), null);
		} catch (IOException e) {
			System.err.println(this + " " + e.getMessage());
		}
	}

	public void run() {
		try {
			String checkPoint = null;
			
			loadProps();
			checkPoint = props.getProperty(inputShard.getShardId());

			GetShardIteratorRequest getShardIteratorRequest = new GetShardIteratorRequest();
			getShardIteratorRequest.setStreamName(inputStreamName);
			getShardIteratorRequest.setShardId(inputShard.getShardId());

			// FIXME: DEBUG
//			checkPoint = null;

			if (checkPoint != null) {
				getShardIteratorRequest.setShardIteratorType("AFTER_SEQUENCE_NUMBER");
				getShardIteratorRequest.setStartingSequenceNumber(checkPoint);
			} else {
				getShardIteratorRequest.setShardIteratorType("TRIM_HORIZON");
			}
			GetShardIteratorResult getShardIteratorResult = amazonKinesisClient.getShardIterator(getShardIteratorRequest);
			String shardIterator = getShardIteratorResult.getShardIterator();

			System.err.println(this.getName() + " Worker started");

			long previous_behind = Long.MAX_VALUE;
			long current_behind = 0l;
			int count = 0;

			while (true) try {

				// Create a new getRecordsRequest with an existing shardIterator 
				GetRecordsRequest getRecordsRequest = new GetRecordsRequest();
				getRecordsRequest.setShardIterator(shardIterator);
				getRecordsRequest.setLimit(inputBatchSize); 

				GetRecordsResult result = null;
				while (result == null) {
					try {
						result = amazonKinesisClient.getRecords(getRecordsRequest);
//						System.err.println(this.getName() + " Received " + result.getRecords().size() + " records");
					} catch (ExpiredIteratorException e) {
//						System.err.println(this.getName() + " Iterator expired, getting new one");
						getShardIteratorResult = amazonKinesisClient.getShardIterator(getShardIteratorRequest);
						shardIterator = getShardIteratorResult.getShardIterator();
						getRecordsRequest = new GetRecordsRequest();
						getRecordsRequest.setShardIterator(shardIterator);
						getRecordsRequest.setLimit(inputBatchSize); 
						result = amazonKinesisClient.getRecords(getRecordsRequest);
//						System.err.println(this.getName() + " Received " + result.getRecords().size() + " records");
					} catch (ProvisionedThroughputExceededException e) {
						System.err.println(this.getName() + " Read throughput exceeded, waiting a turn");
						throttled = true;
						if (threadsleep > 50)
							threadsleep*=1.05;
						else 
							if (inputBatchSize < 20) 
								inputBatchSize++;
							else 
								inputBatchSize = Math.min( maxInputBatchSize, Math.round(inputBatchSize*1.05f) );
						try { Thread.sleep(threadsleep/10); } catch (InterruptedException ee) { }
					}
				}
				// log the reads to cloudwatch
				inputReporter.add(result.getRecords().size());

				long starttime = System.currentTimeMillis();
				for (Record record : result.getRecords()) {
					try { processRecord(record); } catch (Exception e) { System.err.println(this + " " + e.getMessage()); }
					checkPoint = record.getSequenceNumber();
				}
				emitGraphUpdatesToStream();
				long endtime = System.currentTimeMillis();
				
				synchronized(updateConsumer) {
					updateConsumer.totalRecordCount += result.getRecords().size();
					adjustThresholds(updateConsumer.totalRecordCount);
					updateConsumer.saveProps();
				}
				
				// FIXME: DEBUG
//				checkPoint = null;				
				if (checkPoint != null) {
//					System.err.println(this.getName() + " Checkpoint");
					props.setProperty(inputShard.getShardId(), checkPoint);
					saveProps();
					File f = new File(fname+".persistedCounts");
					ObjectOutputStream oos;
					try {
						oos = new ObjectOutputStream(new FileOutputStream(f, false));
						oos.writeObject(dep_table);
						oos.writeObject(l2_table);
						oos.writeObject(DG);
					} catch (FileNotFoundException e) {
						System.err.println(this + " " + e.getMessage());
					} catch (IOException e) {
						System.err.println(this + " " + e.getMessage());
					} catch (Exception e) {
						System.err.println(this.getName() + " " + e.getMessage());
					}
				}
				
				previous_behind = current_behind;
				current_behind = result.getMillisBehindLatest();
				if (!throttled) {
					if (count>0) count--;
					System.err.println(outputString + ", processed " + result.getRecords().size() + "/" + inputBatchSize + " records in " + (endtime-starttime) + " millis at " + current_behind + " millis behind tip, threadsleep " + threadsleep + " count " + count);
				}
				else {
					System.err.println(outputString + ", processed " + result.getRecords().size() + "/" + inputBatchSize + " records in " + (endtime-starttime) + " millis at " + current_behind + " millis behind tip, threadsleep " + threadsleep + " count " + count + " === Throttled ===");
					count=5;
				}

				if ( (result.getRecords().size() == inputBatchSize) && (current_behind > 10000) && (current_behind > previous_behind) && !throttled & count==0)
					// speeding up by shortening our sleep period
					if (threadsleep > 50)
						threadsleep*=0.95;
					else
						if (inputBatchSize < 20) 
							inputBatchSize++;
						else 
							inputBatchSize = Math.min( maxInputBatchSize, Math.round(inputBatchSize*1.05f) );
				if ( result.getRecords().size() < inputBatchSize/2 ) {
					if (threadsleep < 400)
						threadsleep*=1.05;
					else
						inputBatchSize*=0.95;
				}
				throttled=false;

				try { Thread.sleep(threadsleep); } catch (InterruptedException exception) { saveProps(); }

				shardIterator = result.getNextShardIterator();
			} catch (Exception e) { System.err.println(this.getName() + " " + e.getLocalizedMessage()); }	
		} finally {
			saveProps();
		}
	}

}
