/*
 * Copyright 2013 LinkedIn Corp.
 * 
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */

package azkaban.executor;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import azkaban.flow.Node;
import azkaban.utils.Props;
import azkaban.utils.PropsUtils;
import azkaban.utils.TypedMapWrapper;

/**
 * Base Executable that nodes and flows are based.
 */
public class ExecutableNode {
	public static final String ID_PARAM = "id";
	public static final String STATUS_PARAM = "status";
	public static final String STARTTIME_PARAM = "startTime";
	public static final String ENDTIME_PARAM = "endTime";
	public static final String UPDATETIME_PARAM = "updateTime";
	public static final String INNODES_PARAM = "inNodes";
	public static final String OUTNODES_PARAM = "outNodes";
	public static final String TYPE_PARAM = "type";
	public static final String PROPS_SOURCE_PARAM = "propSource";
	public static final String JOB_SOURCE_PARAM = "jobSource";
	public static final String OUTPUT_PROPS_PARAM = "outputProps";
	
	private String id;
	private String type = null;
	private Status status = Status.READY;
	private long startTime = -1;
	private long endTime = -1;
	private long updateTime = -1;
	
	// Path to Job File
	private String jobSource; 
	// Path to top level props file
	private String propsSource;
	private Set<String> inNodes = new HashSet<String>();
	private Set<String> outNodes = new HashSet<String>();
	
	private Props inputProps;
	private Props outputProps;

	public static final String ATTEMPT_PARAM = "attempt";
	public static final String PASTATTEMPTS_PARAM = "pastAttempts";
	
	private int attempt = 0;
	private long delayExecution = 0;
	private ArrayList<ExecutionAttempt> pastAttempts = null;
	
	// Transient. These values aren't saved, but rediscovered.
	private ExecutableFlowBase parentFlow; 
	
	public ExecutableNode(Node node) {
		this.id = node.getId();
		this.jobSource = node.getJobSource();
		this.propsSource = node.getPropsSource();
	}
	
	public ExecutableNode(Node node, ExecutableFlowBase parent) {
		this(node.getId(), node.getType(), node.getJobSource(), node.getPropsSource(), parent);
	}

	public ExecutableNode(String id, String type, String jobSource, String propsSource, ExecutableFlowBase parent) {
		this.id = id;
		this.jobSource = jobSource;
		this.propsSource = propsSource;
		this.type = type;
		setParentFlow(parent);
	}
	
	public ExecutableNode() {
	}
	
	public ExecutableFlow getExecutableFlow() {
		if (parentFlow == null) {
			return null;
		}
		
		return parentFlow.getExecutableFlow();
	}
	
	public void setParentFlow(ExecutableFlowBase flow) {
		this.parentFlow = flow;
	}
	
	public ExecutableFlowBase getParentFlow() {
		return parentFlow;
	}
	
	public String getId() {
		return id;
	}
	
	public void setId(String id) {
		this.id = id;
	}
	
	public Status getStatus() {
		return status;
	}

	public String getType() {
		return type;
	}

	public void setType(String type) {
		this.type = type;
	}
	
	public void setStatus(Status status) {
		this.status = status;
	}
	
	public long getStartTime() {
		return startTime;
	}

	public void setStartTime(long startTime) {
		this.startTime = startTime;
	}

	public long getEndTime() {
		return endTime;
	}

	public void setEndTime(long endTime) {
		this.endTime = endTime;
	}
	
	public long getUpdateTime() {
		return updateTime;
	}

	public void setUpdateTime(long updateTime) {
		this.updateTime = updateTime;
	}
	
	public void addOutNode(String exNode) {
		outNodes.add(exNode);
	}
	
	public void addInNode(String exNode) {
		inNodes.add(exNode);
	}

	public Set<String> getOutNodes() {
		return outNodes;
	}
	
	public Set<String> getInNodes() {
		return inNodes;
	}
	
	public boolean hasJobSource() {
		return jobSource != null;
	}
	
	public boolean hasPropsSource() {
		return propsSource != null;
	}
	
	public String getJobSource() {
		return jobSource;
	}
	
	public String getPropsSource() {
		return propsSource;
	}
	
	public void setInputProps(Props input) {
		this.inputProps = input;
	}
	
	public void setOutputProps(Props output) {
		this.outputProps = output;
	}

	public Props getInputProps() {
		return this.inputProps;
	}
	
	public Props getOutputProps() {
		return outputProps;
	}

	public long getDelayedExecution() {
		return delayExecution;
	}
	
	public void setDelayedExecution(long delayMs) {
		delayExecution = delayMs;
	}
	
	public List<ExecutionAttempt> getPastAttemptList() {
		return pastAttempts;
	}
	
	public int getAttempt() {
		return attempt;
	}

	public void setAttempt(int attempt) {
		this.attempt = attempt;
	}
	
	public void resetForRetry() {
		ExecutionAttempt pastAttempt = new ExecutionAttempt(attempt, this);
		attempt++;
		
		synchronized (this) {
			if (pastAttempts == null) {
				pastAttempts = new ArrayList<ExecutionAttempt>();
			}
			
			pastAttempts.add(pastAttempt);
		}
		
		this.setStartTime(-1);
		this.setEndTime(-1);
		this.setUpdateTime(System.currentTimeMillis());
		this.setStatus(Status.READY);
	}
	
	public List<Object> getAttemptObjects() {
		ArrayList<Object> array = new ArrayList<Object>();
		
		for (ExecutionAttempt attempt: pastAttempts) {
			array.add(attempt.toObject());
		}
		
		return array;
	}
	
	public String getNestedId() {
		return getPrintableId(":");
	}
	
	public String getPrintableId(String delimiter) {
		if (this.getParentFlow() == null || this.getParentFlow() instanceof ExecutableFlow) {
			return getId();
		}
		return getParentFlow().getPrintableId(delimiter) + delimiter + getId();
	}
	
	public Map<String,Object> toObject() {
		Map<String,Object> mapObj = new HashMap<String,Object>();
		fillMapFromExecutable(mapObj);
		
		return mapObj;
	}
	
	protected void fillMapFromExecutable(Map<String,Object> objMap) {
		objMap.put(ID_PARAM, this.id);
		objMap.put(STATUS_PARAM, status.toString());
		objMap.put(STARTTIME_PARAM, startTime);
		objMap.put(ENDTIME_PARAM, endTime);
		objMap.put(UPDATETIME_PARAM, updateTime);
		objMap.put(TYPE_PARAM, type);
		objMap.put(ATTEMPT_PARAM, attempt);
		
		if (inNodes != null && !inNodes.isEmpty()) {
			objMap.put(INNODES_PARAM, inNodes);
		}
		if (outNodes != null && !outNodes.isEmpty()) {
			objMap.put(OUTNODES_PARAM, outNodes);
		}
		
		if (hasPropsSource()) {
			objMap.put(PROPS_SOURCE_PARAM, this.propsSource);
		}
		if (hasJobSource()) {
			objMap.put(JOB_SOURCE_PARAM, this.jobSource);
		}
		
		if (outputProps != null && outputProps.size() > 0) {
			objMap.put(OUTPUT_PROPS_PARAM, PropsUtils.toStringMap(outputProps, true));
		}
		
		if (pastAttempts != null) {
			ArrayList<Object> attemptsList = new ArrayList<Object>(pastAttempts.size());
			for (ExecutionAttempt attempts : pastAttempts) {
				attemptsList.add(attempts.toObject());
			}
			objMap.put(PASTATTEMPTS_PARAM, attemptsList);
		}
	}
	
	@SuppressWarnings("unchecked")
	public void fillExecutableFromMapObject(TypedMapWrapper<String, Object> wrappedMap) {
		this.id = wrappedMap.getString(ID_PARAM);
		this.type = wrappedMap.getString(TYPE_PARAM);
		this.status = Status.valueOf(wrappedMap.getString(STATUS_PARAM));
		this.startTime = wrappedMap.getLong(STARTTIME_PARAM);
		this.endTime = wrappedMap.getLong(ENDTIME_PARAM);
		this.updateTime = wrappedMap.getLong(UPDATETIME_PARAM);
		this.attempt = wrappedMap.getInt(ATTEMPT_PARAM, 0);

		this.inNodes = new HashSet<String>();
		this.inNodes.addAll(wrappedMap.getStringCollection(INNODES_PARAM, Collections.<String>emptySet()));
		
		this.outNodes = new HashSet<String>();
		this.outNodes.addAll(wrappedMap.getStringCollection(OUTNODES_PARAM, Collections.<String>emptySet()));
		
		this.propsSource = wrappedMap.getString(PROPS_SOURCE_PARAM);
		this.jobSource = wrappedMap.getString(JOB_SOURCE_PARAM);
		
		Map<String, String> outputProps = wrappedMap.<String,String>getMap(OUTPUT_PROPS_PARAM);
		if (outputProps != null) {
			this.outputProps = new Props(null, outputProps);
		}
		
		Collection<Object> pastAttempts = wrappedMap.<Object>getCollection(PASTATTEMPTS_PARAM);
		if (pastAttempts!=null) {
			ArrayList<ExecutionAttempt> attempts = new ArrayList<ExecutionAttempt>();
			for (Object attemptObj: pastAttempts) {
				ExecutionAttempt attempt = ExecutionAttempt.fromObject(attemptObj);
				attempts.add(attempt);
			}
			
			this.pastAttempts = attempts;
		}
	}

	public void fillExecutableFromMapObject(Map<String,Object> objMap) {
		TypedMapWrapper<String, Object> wrapper = new TypedMapWrapper<String, Object>(objMap);
		fillExecutableFromMapObject(wrapper);
	}

	public Map<String, Object> toUpdateObject() {
		Map<String, Object> updatedNodeMap = new HashMap<String,Object>();
		updatedNodeMap.put(ID_PARAM, getId());
		updatedNodeMap.put(STATUS_PARAM, getStatus().getNumVal());
		updatedNodeMap.put(STARTTIME_PARAM, getStartTime());
		updatedNodeMap.put(ENDTIME_PARAM, getEndTime());
		updatedNodeMap.put(UPDATETIME_PARAM, getUpdateTime());
		
		updatedNodeMap.put(ATTEMPT_PARAM, getAttempt());

		if (getAttempt() > 0) {
			ArrayList<Map<String,Object>> pastAttempts = new ArrayList<Map<String,Object>>();
			for (ExecutionAttempt attempt: getPastAttemptList()) {
				pastAttempts.add(attempt.toObject());
			}
			updatedNodeMap.put(PASTATTEMPTS_PARAM, pastAttempts);
		}
		
		return updatedNodeMap;
	}
	
	public void applyUpdateObject(TypedMapWrapper<String, Object> updateData) {
		this.status = Status.fromInteger(updateData.getInt(STATUS_PARAM, this.status.getNumVal()));
		this.startTime = updateData.getLong(STARTTIME_PARAM);
		this.updateTime = updateData.getLong(UPDATETIME_PARAM);
		this.endTime = updateData.getLong(ENDTIME_PARAM);
		
		if (updateData.containsKey(ATTEMPT_PARAM)) {
			attempt = updateData.getInt(ATTEMPT_PARAM);
			if (attempt > 0) {
				updatePastAttempts(
						updateData.<Object>getList(PASTATTEMPTS_PARAM, Collections.<Object>emptyList()));
			}
		}
	}

	public void applyUpdateObject(Map<String, Object> updateData) {
		TypedMapWrapper<String,Object> wrapper = new TypedMapWrapper<String,Object>(updateData);
		applyUpdateObject(wrapper);
	}
	
	public void killNode(long killTime) {
		if (this.status == Status.DISABLED) {
			skipNode(killTime);
		}
		else {
			this.setStatus(Status.KILLED);
			this.setStartTime(killTime);
			this.setEndTime(killTime);
			this.setUpdateTime(killTime);
		}
	}
	
	public void skipNode(long skipTime) {
		this.setStatus(Status.SKIPPED);
		this.setStartTime(skipTime);
		this.setEndTime(skipTime);
		this.setUpdateTime(skipTime);
	}
	
	private void updatePastAttempts(List<Object> pastAttemptsList) {
		if (pastAttemptsList == null) {
			return;
		}
		
		synchronized (this) {
			if (this.pastAttempts == null) {
				this.pastAttempts = new ArrayList<ExecutionAttempt>();
			}

			// We just check size because past attempts don't change
			if (pastAttemptsList.size() <= this.pastAttempts.size()) {
				return;
			}

			Object[] pastAttemptArray = pastAttemptsList.toArray();
			for (int i = this.pastAttempts.size(); i < pastAttemptArray.length; ++i) {
				ExecutionAttempt attempt = ExecutionAttempt.fromObject(pastAttemptArray[i]);
				this.pastAttempts.add(attempt);
			}
		}
	}
	
	public int getRetries() {
		return inputProps.getInt("retries", 0);
	}
	
	public long getRetryBackoff() {
		return inputProps.getLong("retry.backoff", 0);
	}
}

