package com.atlassian.stash.internal.repository.sync.auto;

import com.atlassian.bitbucket.concurrent.BucketedExecutor;
import com.atlassian.bitbucket.concurrent.BucketedExecutorSettings;
import com.atlassian.bitbucket.concurrent.ConcurrencyPolicy;
import com.atlassian.bitbucket.concurrent.ConcurrencyService;
import com.atlassian.bitbucket.event.repository.RepositoryRefsChangedEvent;
import com.atlassian.bitbucket.permission.Permission;
import com.atlassian.bitbucket.repository.RefChange;
import com.atlassian.bitbucket.repository.Repository;
import com.atlassian.bitbucket.repository.RepositoryService;
import com.atlassian.bitbucket.repository.sync.RefsSynchronizedEvent;
import com.atlassian.bitbucket.scm.git.GitRefPattern;
import com.atlassian.bitbucket.scm.git.GitScm;
import com.atlassian.bitbucket.user.EscalatedSecurityContext;
import com.atlassian.bitbucket.user.SecurityService;
import com.atlassian.bitbucket.util.*;
import com.atlassian.bitbucket.util.concurrent.ExecutorUtils;
import com.atlassian.event.api.EventListener;
import com.atlassian.stash.internal.repository.sync.InternalRefSyncService;
import com.atlassian.stash.internal.repository.sync.RefSyncConfig;
import com.google.common.base.Predicate;
import com.google.common.collect.Iterables;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

import static com.atlassian.bitbucket.scm.git.GitRefPattern.HEADS;
import static com.atlassian.bitbucket.scm.git.GitRefPattern.TAGS;
import static com.atlassian.bitbucket.util.MoreCollectors.toImmutableList;
import static com.atlassian.bitbucket.util.PageUtils.asPageOf;

/**
 * Listens for {@link RepositoryRefsChangedEvent}s and schedules automatic ref sync processing.
 *
 * @since 3.3
 */
public class AutoRefSyncListener {

    public static final String EXECUTOR_NAME = "ref-sync";

    private static final Function<AutoRefSyncRequest, String> TO_BUCKET_ID = request -> String.valueOf(request.getRepositoryId());

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

    private final RefSyncConfig config;
    private final ExecutorService queueExecutor;
    private final Predicate<Repository> isEnabled = new Predicate<Repository>() {

        @Override
        public boolean apply(Repository repository) {
            return syncService.isEnabled(repository);
        }
    };
    private final RepositoryService repositoryService;
    private final BucketedExecutor<AutoRefSyncRequest> syncExecutor;
    private final InternalRefSyncService syncService;
    private final EscalatedSecurityContext withRepoRead;

    public AutoRefSyncListener(ConcurrencyService concurrencyService, RefSyncConfig config,
                               ExecutorService executorService, RepositoryService repositoryService,
                               SecurityService securityService, AutoRefSyncProcessor syncProcessor,
                               InternalRefSyncService syncService) {
        this.config = config;
        this.repositoryService = repositoryService;
        this.syncService = syncService;

        queueExecutor = executorService;
        syncExecutor = concurrencyService.getBucketedExecutor(EXECUTOR_NAME,
                new BucketedExecutorSettings.Builder<>(TO_BUCKET_ID, syncProcessor)
                        .batchSize(Integer.MAX_VALUE) //Always consume the entire bucket
                        .maxAttempts(1) //No retry; if ref sync fails once it most likely will never succeed
                        .maxConcurrency(config.getThreadCount(), ConcurrencyPolicy.PER_CLUSTER)
                        .build());
        withRepoRead = securityService.withPermission(Permission.REPO_READ, "Retrieving forks for synchronization");
    }

    @EventListener
    public void onRefsChanged(RepositoryRefsChangedEvent event) {
        List<RefChange> refChanges = event.getRefChanges().stream()
                .filter(AutoRefSyncListener::isStandardRef)
                .collect(toImmutableList());

        if (config.isPossible() && isSynchronizable(event, refChanges)) {
            //Use a separate executor to process the event to reduce the delay for returning the event thread
            //to its executor, which can result in dropped events
            queueExecutor.submit(new QueueForksOperation(event, refChanges));
        } else {
            log.debug("Ignoring event; synchronization is not possible");
        }
    }

    /**
     * Shuts down the {@code ExecutorService} which is used to queue processing to avoid doing work on the event
     * processor threads.
     */
    public void shutdown() {
        ExecutorUtils.shutdown(queueExecutor, log);
        syncExecutor.shutdown();
    }

    private static boolean isStandardRef(RefChange refChange) {
        return startsWith(refChange, HEADS) ||
                startsWith(refChange, TAGS);
    }

    private static boolean isSynchronizable(RepositoryRefsChangedEvent event, List<RefChange> refChanges) {
        Repository repository = event.getRepository();
        if (!GitScm.ID.equals(repository.getScmId())) {
            log.trace("{}: Synchronization is only supported for Git repositories", repository);

            return false;
        }

        if (refChanges.isEmpty()) {
            //Logged to trace because events without any changes are normal, expected behaviour
            log.trace("{}: No refs were changed; synchronization is not necessary", repository);

            return false;
        }
        if (event.getUser() == null) {
            log.debug("{}: Not synchronizing {} change(s); there is no associated user", repository, refChanges.size());

            return false;
        }

        return true;
    }

    private static boolean startsWith(RefChange refChange, GitRefPattern refPattern) {
        return refChange.getRef().getId().startsWith(refPattern.getPath());
    }

    private class QueueForksOperation implements PageProvider<Repository>, Runnable, UncheckedOperation<Void> {

        private final AutoRefSyncRequest.Builder builder;
        private final RepositoryRefsChangedEvent event;
        private final Repository repository;

        private QueueForksOperation(RepositoryRefsChangedEvent event, List<RefChange> refChanges) {
            this.event = event;

            builder = new AutoRefSyncRequest.Builder(refChanges, event.getUser());
            repository = event.getRepository();
        }

        @Override
        public Page<Repository> get(PageRequest pageRequest) {
            return asPageOf(Repository.class, repositoryService.findByOrigin(repository, pageRequest));
        }

        @Override
        public void run() {
            try (Timer ignored = TimerUtils.start(repository.getId() + ": Queueing forks for synchronization")) {
                withRepoRead.call(this);
            }

            if (event instanceof RefsSynchronizedEvent) {
                //This event was raised because the refs in the repository were just synced from its upstream. That means
                //any relevant status changes have already been applied
                log.trace("{}: Not updating status after RefsSynchronizedEvent", repository);
            } else if (syncService.isAvailable(repository) && syncService.isEnabled(repository)) {
                //noinspection ConstantConditions
                builder.operation(AutoRefSyncOperation.REVIEW)
                        .repository(repository)
                        .upstream(repository.getOrigin()); //syncService.isAvailable protects this

                //Otherwise, after a push, pull request merge or other RepositoryRefsChangedEvent, if synchronization is
                //both enabled _and_ available, update the synchronization status for all changed refs
                syncExecutor.schedule(builder.build(), 30L, TimeUnit.SECONDS);
            } else {
                //If synchronization is either not enabled or not available, don't update anything. When synchronization
                //is enabled, the up-front full sync will ensure the correct status for everything
                log.debug("{}: Not updating status; synchronization is either not enabled or not available", repository);
            }
        }

        @Override
        public Void perform() {
            builder.operation(AutoRefSyncOperation.SYNCHRONIZE)
                    .upstream(repository);

            for (Repository fork : Iterables.filter(new PagedIterable<>(this, 25), isEnabled)) {
                if (syncService.isAvailable(fork)) {
                    syncExecutor.schedule(builder.repository(fork).build(), config.getSyncDelay(), TimeUnit.SECONDS);
                } else {
                    //This state is now displayed as a banner in the UI when browsing the repository, so it doesn't
                    //need to log very loudly anymore
                    log.debug("{}: Not synchronizing with {}; synchronization is no longer available", fork, repository);
                }
            }
            
            return null;
        }
    }
}
