package geniusweb.protocol.session.learn;

import java.io.IOException;
import java.sql.Date;
import java.text.SimpleDateFormat;
import java.util.Collections;
import java.util.List;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.logging.Level;
import java.util.stream.Collectors;

import geniusweb.actions.Action;
import geniusweb.actions.PartyId;
import geniusweb.deadline.Deadline;
import geniusweb.events.ProtocolEvent;
import geniusweb.inform.Finished;
import geniusweb.inform.Inform;
import geniusweb.inform.Settings;
import geniusweb.progress.ProgressFactory;
import geniusweb.protocol.CurrentNegoState;
import geniusweb.protocol.NegoState;
import geniusweb.protocol.ProtocolException;
import geniusweb.protocol.partyconnection.ProtocolToPartyConn;
import geniusweb.protocol.partyconnection.ProtocolToPartyConnFactory;
import geniusweb.protocol.partyconnection.ProtocolToPartyConnections;
import geniusweb.protocol.session.SessionProtocol;
import geniusweb.references.Parameters;
import geniusweb.references.PartyWithProfile;
import geniusweb.references.ProfileRef;
import geniusweb.references.ProtocolRef;
import geniusweb.references.Reference;
import tudelft.utilities.listener.DefaultListenable;
import tudelft.utilities.logging.Reporter;
import tudelft.utilities.repository.NoResourcesNowException;

/**
 * The Learn protocol allows parties to learn until the deadline set in the
 * {@link LearnSettings}.
 */
public class Learn extends DefaultListenable<ProtocolEvent>
		implements SessionProtocol {
	public static final int MIN_SLEEP_TIME = 1000;
	public static final int MAX_SLEEP_TIME = 60000;
	public static final int TIME_MARGIN = 20;// ms extra delay after deadline
	private LearnState state;
	private final Reporter log;
	private static final ProtocolRef LEARN = new ProtocolRef("Learn");
	private volatile Timer deadlinetimer = null;
	private volatile AtomicBoolean isFinishedInfoSent = new AtomicBoolean(
			false);
	private volatile ProtocolToPartyConnections conns = new ProtocolToPartyConnections(
			Collections.emptyList());

	public Learn(LearnState state, Reporter logger) {
		this.state = state;
		this.log = logger;
	}

	@Override
	public void start(ProtocolToPartyConnFactory connectionfactory) {
		try {
			connect(connectionfactory);
			setDeadline();
			setupParties();
		} catch (Throwable e) {
			handleError("Failed to start up session", null, e);
		}

	}

	@Override
	public String getDescription() {
		return "Sends all parties the Settings. Parties can start learning immediately but must respect the deadline. When party is done, it should send LearningDone. ";
	}

	@Override
	public NegoState getState() {
		return state;
	}

	@Override
	public ProtocolRef getRef() {
		return LEARN;
	}

	@Override
	public void addParticipant(PartyWithProfile party) {
		throw new IllegalStateException(
				"Dynamic joining a negotiation is not allowed in LEARN");
	}

	/*******************************************************************
	 * private functions. Some are protected only, for testing purposes
	 ********************************************************************/
	/**
	 * step 1 in protocol: connect all involved parties and start the clock.
	 * This always "succeeds" with a valid (but possibly final) state
	 * <p>
	 * This is 'protected' to allow junit testing, this code is not a 'public'
	 * part of the interface.
	 * 
	 * @param connectionfactory the connectionfactory for making party
	 *                          connections
	 * 
	 * @throws InterruptedException if the connection procedure is unterrupted
	 * 
	 * @throws IOException          if this fails to properly conect to the
	 *                              parties, eg interrupted or server not
	 *                              responding..
	 */
	protected synchronized void connect(
			ProtocolToPartyConnFactory connectionfactory)
			throws InterruptedException, IOException {
		List<PartyWithProfile> participants = state.getSettings()
				.getAllParties();
		List<Reference> parties = participants.stream()
				.map(parti -> (parti.getParty().getPartyRef()))
				.collect(Collectors.toList());
		List<ProtocolToPartyConn> connections = null;
		log.log(Level.INFO, "LEARN connect " + parties);
		while (connections == null) {
			try {
				connections = connectionfactory.connect(parties);
			} catch (NoResourcesNowException e) {
				long waitms = e.getLater().getTime()
						- System.currentTimeMillis();
				log.log(Level.INFO,
						"No resources available to run session, waiting"
								+ waitms);
				Thread.sleep(Math.min(MAX_SLEEP_TIME,
						Math.max(MIN_SLEEP_TIME, waitms)));
			}
		}
		for (int i = 0; i < participants.size(); i++) {
			conns = conns.with(connections.get(i));
			setState(this.state.with(connections.get(i).getParty(),
					participants.get(i)));
		}
	}

	/**
	 * This is called when one of the party connections does an action.
	 * Synchronized so that we always handle only 1 action at a time.
	 * 
	 * @param partyconn the connection on which the action came in.
	 * @param action    the {@link Action} taken by some party
	 */
	protected synchronized void actionRequest(
			final ProtocolToPartyConn partyconn, final Action action) {
		if (action == null) {
			Throwable err = partyconn.getError();
			if (err == null) {
				err = new ProtocolException("Party sent a null action",
						partyconn.getParty());
			}
			handleError(partyconn + "Protocol error", partyconn.getParty(),
					err);
			return;
		}

		try {
			setState(state.with(partyconn.getParty(), action));
		} catch (Throwable e) {
			handleError("failed to handle action " + action,
					partyconn.getParty(), e);
		}

	}

	/**
	 * step 2 in protocol: listen to connections and send settings to the
	 * parties.
	 * <p>
	 * This is 'protected' to allow junit testing, this code is not a 'public'
	 * part of the interface.
	 * 
	 * @throws ProtocolException if a party does not follow the protocol
	 */
	protected synchronized void setupParties() throws ProtocolException {
		for (ProtocolToPartyConn conn : conns) {
			conn.addListener(action -> actionRequest(conn, action));
		}

		for (ProtocolToPartyConn connection : conns) {
			try {
				sendSettings(connection);
			} catch (IOException e) {
				throw new ProtocolException("Failed to initialize",
						connection.getParty(), e);
			}
		}

	}

	/**
	 * Inform a party about its settings
	 * 
	 * @param connection
	 * @throws IOException if party got disconnected
	 */
	private synchronized void sendSettings(ProtocolToPartyConn connection)
			throws IOException {
		PartyId partyid = connection.getParty();
		ProfileRef profile = state.getPartyProfiles().get(partyid).getProfile();
		Parameters params = state.getPartyProfiles().get(partyid).getParty()
				.getParameters();
		if (profile == null) {
			throw new IllegalArgumentException(
					"Missing profile for party " + connection.getReference());
		}
		connection.send(new Settings(connection.getParty(), profile, getRef(),
				state.getProgress(), params));
	}

	/**
	 * Set state to proper deadline. Starts the timer tasks. This tasks triggers
	 * a call to handleError when the session times out.
	 */
	private synchronized void setDeadline() {
		long now = System.currentTimeMillis();
		Deadline deadline = state.getSettings().getDeadline();
		setState(state.with(ProgressFactory.create(deadline, now)));
		deadlinetimer = new Timer();
		TimerTask task = new TimerTask() {
			@Override
			public void run() {
				if (!state.isFinal(System.currentTimeMillis())) {
					log.log(Level.SEVERE,
							"BUG. Deadline timer has triggered but state is not final");
				}
				log.log(Level.INFO,
						"LEARN deadline reached. Terminating session.");
				finish();
			}
		};
		// set timer TIME_MARGIN after real deadline to ensure we're not too
		// early
		deadlinetimer.schedule(task, deadline.getDuration() + TIME_MARGIN);
		log.log(Level.INFO, "LEARN deadline set to "
				+ new SimpleDateFormat("yyyy/MM/dd HH:mm:ss")
						.format(new Date(System.currentTimeMillis()
								+ deadline.getDuration() + TIME_MARGIN)));
	}

	/**
	 * Update state to include the given error and finishes up the session.
	 * 
	 * @param message The message to attach to the error
	 * @param party   the party where the error occured
	 * @param e       the exception that occured.
	 */
	private synchronized void handleError(final String message,
			final PartyId party, final Throwable e) {
		if (e instanceof ProtocolException) {
			setState(state.with((ProtocolException) e));
		} else {
			setState(state.with(new ProtocolException(message, party, e)));
		}
		log.log(Level.WARNING, "LEARN protocol intercepted error due to party "
				+ party + ":" + message, e);
	}

	/**
	 * Sets the new state. If the new state is final, the finish-up procedure is
	 * executed.
	 * 
	 * @param newstate the new state.
	 */
	private synchronized void setState(LearnState newstate) {
		long now = System.currentTimeMillis();
		if (state.isFinal(now)) {
			finish();
			return;
		}
		this.state = newstate;
		if (newstate.isFinal(now)) {
			finish();
		}
	}

	/**
	 * Called when we reach final state. Cancels deadline timer. Send finished
	 * info to all parties, notify current nego state as final and set
	 * {@link #isFinishedInfoSent}. Double calls are automatically ignored.
	 */
	private synchronized void finish() {
		if (deadlinetimer != null) {
			deadlinetimer.cancel();
			deadlinetimer = null;
		}
		if (!isFinishedInfoSent.compareAndSet(false, true))
			return;
		Inform finished = new Finished(state.getAgreements());
		for (ProtocolToPartyConn conn : conns) {
			sendFinish(conn, finished);
		}
		notifyListeners(new CurrentNegoState(state));
	}

	private void sendFinish(ProtocolToPartyConn connection, Inform finished) {
		try {
			connection.send(finished);
			connection.close();
		} catch (Exception e) {
			log.log(Level.INFO, "Failed to send Finished to " + connection, e);
		}
	}
}