package com.atlassian.stash.internal.throttle;

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

/**
 * {@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>the {@link #tryAcquire()} or {@link #tryAcquire(long, TimeUnit)} times out</li>
 * </ul>
 *
 * @see SemaphoreThrottleService
 * @since 1.0.1
 */
public class SemaphoreTicketBucket {

    private final long defaultAcquireTimeoutMillis;
    private final int permits;
    private final Semaphore semaphore;
    private final ConcurrentSkipListSet<Timing> timings;

    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 unit                  the unit of {@code defaultAcquireTimeout}
     */
    public SemaphoreTicketBucket(int permits, long defaultAcquireTimeout, @Nonnull TimeUnit unit) {
        this.defaultAcquireTimeoutMillis = Math.max(0L, unit.toMillis(defaultAcquireTimeout));
        this.permits = permits;
        this.semaphore = new Semaphore(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();
    }

    /**
     * Creates a {@link TicketSummary} for this bucket.
     *
     * @param name the name of the bucket
     * @return a summary for this bucket
     * @since 3.4
     */
    public TicketSummary summarize(String name) {
        long oldest = getEarliestQueuingTime();
        long lastRejection = getLastRejectedTimestamp();
        int availablePermits = semaphore.availablePermits();
        return new TicketSummary(availablePermits, lastRejection == 0L ? null : new Date(lastRejection),
                name, oldest == 0L ? null : new Date(oldest), semaphore.getQueueLength(), permits,
                permits - availablePermits);
    }

    /**
     * @return the number of tickets available and the total number of tickets
     */
    @Override
    public String toString() {
        return semaphore.availablePermits() + "/" + 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}
     */
    public boolean tryAcquire() {
        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}
     */
    public boolean tryAcquire(long timeout, TimeUnit timeUnit) {
        boolean acquired = false;

        Timing timing = new Timing();
        timings.add(timing);
        try {
            acquired = semaphore.tryAcquire(timeout, timeUnit);
        } catch (InterruptedException e) {
            // no-op
        } finally {
            timings.remove(timing);
        }
        if (!acquired) {
            lastRejectedTimestamp = System.currentTimeMillis();
        }

        return acquired;
    }

    private static class Timing implements Comparable<Timing> {

        private final long start = System.currentTimeMillis();

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