package com.atlassian.bitbucket.mesh.repair;

import com.atlassian.bitbucket.mesh.repository.Repository;
import com.google.common.collect.Sets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import java.time.Duration;
import java.util.Set;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import static java.util.Objects.requireNonNull;

/**
 * Component that restricts concurrent repair operations:
 * <ul>
 *     <li>For a given repository, only a single repair may be active.</li>
 *     <li>At most {@code maxConcurrency} repairs may be active concurrently.</li>
 * </ul>
 */
public class RepairGate {

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

    private final Semaphore repairSlots;
    private final Set<String> underRepair;

    public RepairGate(int maxConcurrency) {
        repairSlots = new Semaphore(maxConcurrency);
        underRepair = Sets.newConcurrentHashSet();
    }

    /**
     * Attempts to acquire a repair 'ticket' for the provided repository. If a ticket is returned, the caller
     * <em>MUST</em> call {@link RepairTicket#close()} after the repair completes.
     *
     * @param repositoryId the repository for which the repair is requested
     * @param timeout the maximum amount of time to wait for a repair slot to become available
     * @return the {@link RepairTicket}
     * @throws RepositoryRepairAlreadyRunningException if the ticket for the repository has already been acquired
     * @throws RepositoryRepairSlotsExhaustedException if the maximum number of concurrent repairs is already reached,
     *                                                 even after waiting {@code timeout}.
     */
    @Nonnull
    public RepairTicket acquireTicket(@Nonnull String repositoryId, @Nonnull Duration timeout) {
        requireNonNull(repositoryId, "repositoryId");
        requireNonNull(timeout, "timeout");

        repositoryId = Repository.canonicalId(repositoryId);
        if (underRepair.contains(repositoryId)) {
            //Already started
            throw new RepositoryRepairAlreadyRunningException(repositoryId);
        }

        try {
            if (repairSlots.tryAcquire(timeout.toMillis(), TimeUnit.MILLISECONDS)) {
                if (underRepair.add(repositoryId)) {
                    log.debug("Claimed slot for {} (available = {}, underRepair = {})", repositoryId,
                            repairSlots.availablePermits(), underRepair);
                    return new DefaultRepairTicket(repositoryId);
                } else {
                    //Claimed by another thread while we were waiting to acquire a repair slot
                    repairSlots.release();
                    throw new RepositoryRepairAlreadyRunningException(repositoryId);
                }
            }
        } catch (InterruptedException e) {
            //Restore the interrupted flag and throw the RepairSlotsExhaustedException below
            Thread.currentThread().interrupt();
        }

        log.warn("Failed to claim repair slot for {} (available = {}, claims = {})", repositoryId,
                repairSlots.availablePermits(), underRepair);
        throw new RepositoryRepairSlotsExhaustedException(repositoryId);
    }

    /**
     * @return the number of currently active repair processes
     */
    public int getActiveCount() {
        return underRepair.size();
    }

    private class DefaultRepairTicket implements RepairTicket {

        private final String repositoryId;
        private final AtomicBoolean closed;

        DefaultRepairTicket(String repositoryId) {
            this.repositoryId = repositoryId;
            closed = new AtomicBoolean();
        }

        @Override
        public void close() {
            if (closed.compareAndSet(false, true)) {
                underRepair.remove(repositoryId);
                repairSlots.release();
                log.debug("Released slot for {} (available = {}, underRepair = {})", repositoryId,
                        repairSlots.availablePermits(), underRepair);
            }
        }
    }
}
