package com.atlassian.bitbucket.mesh.coordination;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.atomic.AtomicLong;

import static java.lang.Math.max;

/**
 * A simple implementation of a Lamport clock, which is a <em>logical</em> clock that allows us to determine whether
 * one event happened before another event.
 * <p>
 * The clock is updates using the following rules:
 * <ul>
 *     <li>When node A receives a message from node B that contains a timestamp, {@link #onReceived(long)} MUST be
 *     called with the remote node's reported timestamp. Internally, this updates the clock to max(localTimestamp,
 *     remoteTimestamp) + 1.</li>
 *     <li>When node A sends a message that includes a timestamp to one or more other nodes, it MUST call
 *     {@link #tick()} to progress and obtain timestamp for the outgoing message. If the outgoing message is
 *     conceptually a multicast (i.e. the same event being sent to multiple nodes), {@link #tick()} should be called
 *     once and the resulting timestamp used for each outgoing message. Sending a PREPARE or COMMIT instruction
 *     to each replica is a good example of this.</li>
 *     <li>The application <em>SHOULD</em> also call {@link #tick()} to increment the counter when executing an
 *     internal event if the timestamp of that event is somehow relevant (e.g. it should be compared to another
 *     event's timestamp).</li>
 * </ul>
 * Care must be taken when comparing Lamport timestamps from different nodes because they can only be used to create
 * <em>partial causal ordering</em>; if event A caused or influenced event B, (happened-before) timestamp A MUST be
 * before timestamp B.
 * The reverse cannot be concluded though; if timestamp A &lt; timestamp B, A <em>MAY</em> have happened before B,
 * but may also incomparable to B. We can however conclude that B cannot have influenced A.
 * <p>
 * Note: The Lamport clock is internal to Mesh and only used in Mesh-Mesh communication.
 * <p>
 * See <a href="https://en.wikipedia.org/wiki/Lamport_timestamp">Lamport timestamp</a> for more background.
 */
public class LamportClock implements LogicalClock {

    private static final Logger log = LoggerFactory.getLogger(LamportClock.class);

    private final AtomicLong counter;

    public LamportClock() {
        this(0L);
    }

    public LamportClock(long startValue) {
        counter = new AtomicLong(startValue);
    }

    /**
     * @return the current timestamp
     */
    @Override
    public long get() {
        return counter.get();
    }

    /**
     * Synchronizes the local clock with the remote timestamp if the local timestamp is lower than the remote
     * timestamp, and then increases the timestamp by 1 to reflect that a message has been received.
     *
     * @param remoteTimestamp the Lamport timestamp reported by a remote node
     * @return the (possibly) updated timestamp.
     */
    @Override
    public long onReceived(long remoteTimestamp) {
        return counter.updateAndGet(current -> handleOverflow(max(remoteTimestamp, current) + 1L));
    }

    /**
     * Increases the internal event counter by 1 and returns the resulting timestamp
     * @return the updated timestamp
     */
    @Override
    public long tick() {
        return counter.updateAndGet(current -> handleOverflow(current + 1));
    }

    private static long handleOverflow(long newValue) {
        if (newValue < 0L) {
            //It seems unlikely that we'd actually hit this, but let's at least log something.
            //Back of the envelope math suggests that we have 2^63 ticks at our disposal. If we aim high and go
            //for 200 years of application lifetime, we'd have 1.46 billion ticks per second at our disposal.
            //Even assuming a 10k node cluster, that would leave us 146k ticks per second per node.
            log.warn("Clock overflow detected! Restarting the clock..");
            return 0L;
        }
        return newValue;
    }
}
