package geniusweb.protocol.session.saop;

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.ActionDone;
import geniusweb.inform.Finished;
import geniusweb.inform.Inform;
import geniusweb.inform.Settings;
import geniusweb.inform.YourTurn;
import geniusweb.progress.ProgressFactory;
import geniusweb.protocol.CurrentNegoState;
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.protocol.session.SessionSettings;
import geniusweb.protocol.session.SessionState;
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 protocol runs as follows
 * <ol>
 * <li>The protocol tries to start all parties. If not all parties start, the
 * parties are freed up and another attempt is done to start all parties some
 * time later.
 * <li>All parties are sent the {@link SessionSettings}. Only parties specified
 * initially in the settings do participate.
 * <li>The session deadline clock now starts ticking.
 * <li>All parties are sent their settings.
 * <li>All parties get YourTurn in clockwise order. A party must do exactly one
 * action after it received YourTurn.
 * <li>The negotiation continues until an agreement is reached (all parties
 * agreed to the last bid), the {@link Deadline} is reached, or a party fails to
 * adhere to the protocol.
 * <li>If the session times out, the connections are cut and the negotiation
 * completes without errors and without agreement.
 * </ol>
 * <p>
 * This logs to "Protocol" logger if there are issues
 * <p>
 * This object is mutable: the internal state changes as parties interact with
 * the protocol.
 * <p>
 * Thread safe: all entry points are synchronized.
 */
public class SAOP extends DefaultListenable<ProtocolEvent>
		implements SessionProtocol {
	public static final int TIME_MARGIN = 20;// ms extra delay after deadline
	public static final int MINDURATION = 100;
	public static final int MIN_SLEEP_TIME = 1000;
	public static final int MAX_SLEEP_TIME = 60000;
	private static final ProtocolRef SAOP = new ProtocolRef("SAOP");
	private final Reporter log;

	// these 4 variables are why protocols are so complex.
	private SAOPState state = null; // mutable!
	private volatile AtomicBoolean isFinishedInfoSent = new AtomicBoolean(
			false);
	private volatile Timer deadlinetimer = null;
	private volatile ProtocolToPartyConnections conns;

	public SAOP(SAOPState state, Reporter logger,
			ProtocolToPartyConnections connects) {
		if (state == null) {
			throw new NullPointerException("state must be not null");
		}
		if (state.getSettings().getDeadline().getDuration() < MINDURATION) {
			throw new IllegalArgumentException(
					"Duration must be at least " + MINDURATION);
		}
		this.log = logger;
		this.state = state;
		this.conns = connects;
	}

	/**
	 * 
	 * @param state  normally the initial state coming from SAOPSettings
	 * @param logger the Reporter to log to
	 */
	public SAOP(SAOPState state, Reporter logger) {
		this(state, logger,
				new ProtocolToPartyConnections(Collections.emptyList()));
	}

	@Override
	public synchronized void start(
			ProtocolToPartyConnFactory connectionfactory) {

		try {
			connect(connectionfactory);
			setDeadline();
			setupParties();
			nextTurn();
		} catch (Throwable e) {
			handleError("Failed to start up session", null, e);
		}
	}

	@Override
	public String getDescription() {
		return "All parties get YourTurn in clockwise order, after which they can do their next action. "
				+ "No new participants after start. End after prescribed deadline or when some bid is unanimously Accepted."
				+ "Parties can only act on their own behalf and only when it is their turn.";
	}

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

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

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

	/*******************************************************************
	 * 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, "SAOP 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++) {
			// now we bookkeep the connections ourselves,
			// and update the state to keep in sync.
			conns = conns.with(connections.get(i));
			setState(this.state.with(connections.get(i).getParty(),
					participants.get(i)));
		}
	}

	/**
	 * 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,
						"SAOP 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, "SAOP deadline set to "
				+ new SimpleDateFormat("yyyy/MM/dd HH:mm:ss")
						.format(new Date(System.currentTimeMillis()
								+ deadline.getDuration() + TIME_MARGIN)));
	}

	/**
	 * 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));
	}

	/**
	 * 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 {
			if (!partyconn.getParty().equals(state.getNextActor())) {
				// party does not have the turn.
				throw new ProtocolException(
						"Party acts without having the turn",
						partyconn.getParty());
			}
			// FIXME? this ignores possible broadcast errors
			conns.broadcast(new ActionDone(action));
			setState(state.with(partyconn.getParty(), action));
			if (!state.isFinal(System.currentTimeMillis()))
				nextTurn();
		} catch (Throwable e) {
			handleError("failed to handle action " + action,
					partyconn.getParty(), e);
		}

	}

	/**
	 * Signal next participant it's his turn
	 * 
	 * @throws IOException
	 */
	synchronized void nextTurn() {
		PartyId party = state.getNextActor();
		try {
			conns.get(party).send(new YourTurn());
		} catch (IOException e) {
			handleError("failed to send YourTurn", party, e);
		}
	}

	/**
	 * 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. Can be null in
	 *                exceptional cases.
	 * @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)));
		}
		if (party != null)
			setState(state.withoutParty(party));
		log.log(Level.WARNING, "SAOP 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(SAOPState 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);
		}
	}

}