package com.atlassian.bitbucket.mesh.throttle;

import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;

import javax.annotation.Nonnull;
import java.time.Instant;
import java.util.NoSuchElementException;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

import static com.google.common.base.Preconditions.checkArgument;
import static java.util.Objects.requireNonNull;
import static org.slf4j.LoggerFactory.getLogger;

/**
 * {@code SemaphoreTicketBucket} is used to throttle access to expensive operations. A {@code SemaphoreTicketBucket} is
 * associated with one such resource and 'holds' a limited set of tickets. If all available tickets have been required
 * and another ticket is requested, the requesting thread will block until:
 * <ul>
 * <li>a ticket is released and becomes available, or</li>
 * <li>{@link #tryAcquire()} or {@link #tryAcquire(long, TimeUnit)} times out</li>
 * </ul>
 *
 * @see SemaphoreThrottleService
 */
public class SemaphoreTicketBucket {

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

    private final ThrottledResource resource;
    private final ResizeableSemaphore semaphore;
    private final ConcurrentSkipListSet<Timing> timings;

    private volatile long defaultAcquireTimeoutMillis;
    private volatile int permits;
    private volatile long lastRejectedTimestamp;

    /**
     * Constructs a new {@code SemaphoreTicketBucket} with a semaphore allowing the specified number of permits and
     * the provided default acquire timeout.
     *
     * @param permits               the number of tickets that are available
     * @param defaultAcquireTimeout the default timeout for acquiring tickets, used by {@link #tryAcquire()}
     * @param resource the resource that needs to be protected
     * @param unit                  the unit of {@code defaultAcquireTimeout}
     */
    public SemaphoreTicketBucket(@Nonnull ThrottledResource resource, int permits, long defaultAcquireTimeout,
                                 @Nonnull TimeUnit unit) {
        this.permits = permits;
        this.resource = requireNonNull(resource, "resource");

        defaultAcquireTimeoutMillis = Math.max(0L, unit.toMillis(defaultAcquireTimeout));
        semaphore = new ResizeableSemaphore(permits, defaultAcquireTimeout > 0L);
        timings = new ConcurrentSkipListSet<>();
    }

    /**
     * Get the earliest request time of a currently queued request for a ticket
     * @return System time of the earliest queued request for a ticket or  {@code 0} if there is none
     */
    public long getEarliestQueuingTime() {
        if (timings.isEmpty()) {
            return 0L;
        }
        try {
            return timings.first().start;
        } catch (NoSuchElementException e) {
            return 0L;
        }
    }

    /**
     * Retrieves the timestamp of the most recent rejected attempt to acquire a ticket. If no tickets have been
     * rejected, the returned timestamp is {@code 0}.
     *
     * @return the timestamp of the last time an attempt to acquire a ticket was rejected, or {@code 0} if no attempt
     *         has failed
     */
    public long getLastRejectedTimestamp() {
        return lastRejectedTimestamp;
    }

    /**
     * Retrieves the <i>total number</i> of tickets available in this bucket, some of which may currently be in use.
     *
     * @return the number of tickets in the bucket
     */
    public int getTotalTickets() {
        return permits;
    }

    /**
     * Releases a ticket.
     */
    public void release() {
        semaphore.release();
    }

    /**
     * Updates the default timeout for acquiring tickets, used by {@link #tryAcquire()}
     *
     * @param acquireTimeoutMillis the timeout
     */
    public void setAcquireTimeoutMillis(long acquireTimeoutMillis) {
        if (defaultAcquireTimeoutMillis == acquireTimeoutMillis) {
            return;
        }

        synchronized (this) {
            long currentTimeout = defaultAcquireTimeoutMillis;
            if (currentTimeout == acquireTimeoutMillis) {
                return;
            }

            log.trace("{} acquire timeout by {} ms", currentTimeout > acquireTimeoutMillis ? "Reducing" : "Increasing",
                    Math.abs(currentTimeout - defaultAcquireTimeoutMillis));
            defaultAcquireTimeoutMillis = acquireTimeoutMillis;
        }
    }

    /**
     * Creates a {@link TicketSummary} for this bucket.
     *
     * @return a summary for this bucket
     */
    public TicketSummary summarize() {
        long oldest = getEarliestQueuingTime();
        long lastRejection = getLastRejectedTimestamp();
        int availablePermits = semaphore.availablePermits();
        return new TicketSummary(resource, availablePermits,
                lastRejection == 0L ? null : Instant.ofEpochMilli(lastRejection),
                oldest == 0L ? null : Instant.ofEpochMilli(oldest), semaphore.getQueueLength(), permits,
                permits - availablePermits);
    }

    public void setPermits(int newPermits) {
        checkArgument(newPermits >= 0, "Permits must be non-negative");

        if (newPermits == permits) {
            return;
        }
        synchronized (this) {
            int currentPermits = permits;

            if (newPermits == currentPermits) {
                return;
            }

            if (newPermits > currentPermits) {
                int delta = newPermits - currentPermits;
                log.trace("Increasing tickets by {}", delta);
                semaphore.release(delta);
            } else {
                int delta = currentPermits - newPermits;
                log.trace("Decreasing tickets by {}", delta);
                semaphore.reducePermits(delta);
            }
            permits = newPermits;
        }
    }

    /**
     * @return the number of tickets available and the total number of tickets
     */
    @Override
    public String toString() {
        int available = semaphore.availablePermits();
        return available < 0
            ? "0/" + permits + "+" + (-1 * available)
            : available + "/" + permits;
    }

    /**
     * Attempts to acquire a ticket. The calling thread will block for at most {@code defaultAcquireTimeout}.
     *
     * @return {@code true} if the ticket was acquired. Otherwise {@code false}
     * @throws InterruptedException if the thread was interrupted
     */
    public boolean tryAcquire() throws InterruptedException {
        return tryAcquire(defaultAcquireTimeoutMillis, TimeUnit.MILLISECONDS);
    }

    /**
     * Attempts to acquire a ticket. The calling thread will block for at most {@code timeout}.
     *
     * @param timeout  the maximum time to wait for a ticket to become available
     * @param timeUnit the time unit
     * @return {@code true} if the ticket was acquired. Otherwise {@code false}
     * @throws InterruptedException if the thread was interrupted
     */
    public boolean tryAcquire(long timeout, TimeUnit timeUnit) throws InterruptedException {
        Timing timing = new Timing();
        timings.add(timing);
        try {
            if (semaphore.tryAcquire(timeout, timeUnit)) {
                return true;
            }

            lastRejectedTimestamp = System.currentTimeMillis();
            return false;
        } finally {
            timings.remove(timing);
        }
    }

    @VisibleForTesting
    long getDefaultAcquireTimeoutMillis() {
        return defaultAcquireTimeoutMillis;
    }

    private static class Timing implements Comparable<Timing> {

        private final long start = System.currentTimeMillis();

        @Override
        public int compareTo(@Nonnull Timing o) {
            //the gap here should not be longer than max int
            return Long.compare(start, o.start);
        }
    }

    private static class ResizeableSemaphore extends Semaphore {

        private ResizeableSemaphore(int permits, boolean fair) {
            super(permits, fair);
        }

        @Override
        public void reducePermits(int reduction) {
            super.reducePermits(reduction);
        }
    }
}
