package com.atlassian.bitbucket.internal.mirroring.mirror.sync;

import com.atlassian.bitbucket.ServiceException;
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.internal.mirroring.mirror.*;
import com.atlassian.bitbucket.internal.mirroring.mirror.client.InternalUpstreamClient;
import com.atlassian.bitbucket.internal.mirroring.mirror.client.InternalUpstreamClientFactory;
import com.atlassian.bitbucket.internal.mirroring.mirror.dao.AoProjectMapping;
import com.atlassian.bitbucket.internal.mirroring.mirror.dao.AoRepositoryMapping;
import com.atlassian.bitbucket.internal.mirroring.mirror.dao.ProjectMappingDao;
import com.atlassian.bitbucket.internal.mirroring.mirror.dao.RepositoryMappingDao;
import com.atlassian.bitbucket.internal.mirroring.repositories.RepositoryContentHashService;
import com.atlassian.bitbucket.mirroring.mirror.*;
import com.atlassian.bitbucket.project.Project;
import com.atlassian.bitbucket.project.ProjectCreateRequest;
import com.atlassian.bitbucket.project.ProjectService;
import com.atlassian.bitbucket.project.ProjectUpdateRequest;
import com.atlassian.bitbucket.repository.*;
import com.atlassian.bitbucket.scm.ScmFeature;
import com.atlassian.bitbucket.scm.ScmService;
import com.atlassian.bitbucket.scm.UnavailableScmException;
import com.atlassian.bitbucket.scm.UnsupportedScmException;
import com.atlassian.bitbucket.scm.mirror.MirrorUpdateRefsCommandParameters;
import com.atlassian.bitbucket.scm.mirror.ScmMirrorCommandFactory;
import com.atlassian.bitbucket.util.*;
import com.atlassian.bitbucket.util.Timer;
import com.atlassian.event.api.EventPublisher;
import com.atlassian.sal.api.lifecycle.LifecycleAware;
import com.atlassian.sal.api.transaction.TransactionCallback;
import com.atlassian.sal.api.transaction.TransactionTemplate;
import com.google.common.collect.Sets;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.function.Consumer;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static com.atlassian.bitbucket.internal.mirroring.mirror.MirrorDescriptionUtils.describe;
import static com.atlassian.bitbucket.mirroring.mirror.UpstreamServerType.BITBUCKET_SERVER;
import static com.atlassian.bitbucket.util.MoreStreams.streamIterable;
import static com.google.common.base.Preconditions.checkNotNull;
import static java.util.Arrays.asList;
import static java.util.Optional.empty;
import static java.util.Optional.of;
import static java.util.stream.Collectors.toSet;

/**
 * Note: all public methods on this helper class are assumed to be called only by the upstream service user. The
 * {@link DefaultUpstreamService} enforces this.
 */
public class DefaultMirrorSynchronizationHelper implements MirrorSynchronizationHelper, LifecycleAware {

    private static final int REPO_MAPPING_PAGE_MAX = 500;
    private static final String SYNC_EXECUTOR_NAME = "repository-mirror-sync";
    private static final String TIMER_FULL_SYNCHRONIZATION = "void " + FullSynchronizationTask.class.getName() + ".run()";
    private static final String TIMER_REPOSITORY_SYNCHRONIZATION = "void " + RepositorySynchronizationTask.class.getName() + ".run()";
    private static final String TIMER_SYNC_LOCAL_REPOSITORY = "Repository " +
            DefaultMirrorSynchronizationHelper.class.getName() + ".syncLocalRepository(ExternalRepository, SyncContext)";

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

    private final RepositoryContentHashService contentHashService;
    private final DedupingExecutor dedupingExecutor;
    private final EventPublisher eventPublisher;
    private final ExecutorService executorService;
    private final int maxFailedSyncCount;
    private final ProjectMappingDao projectMappingDao;
    private final ProjectService projectService;
    private final RepositoryMappingDao repoMappingDao;
    private final RepositoryService repositoryService;
    private final BucketedExecutor<RepositoryFetchTask> repositoryFetchExecutor;
    private final ScmService scmService;
    private final TransactionTemplate transactionTemplate;
    private final InternalUpstreamClientFactory upstreamClientFactory;
    private final UpstreamSettingsService upstreamSettingsService;
    private final MirroringValidationHelper validationHelper;

    public DefaultMirrorSynchronizationHelper(ConcurrencyService concurrencyService,
                                              MirroringConfig config,
                                              RepositoryContentHashService contentHashService,
                                              EventPublisher eventPublisher,
                                              ExecutorService executorService,
                                              ProjectMappingDao projectMappingDao, ProjectService projectService,
                                              RepositoryFetchProcessor repositoryFetchProcessor,
                                              RepositoryMappingDao repoMappingDao, RepositoryService repositoryService,
                                              ScmService scmService,
                                              TransactionTemplate transactionTemplate,
                                              InternalUpstreamClientFactory upstreamClientFactory,
                                              UpstreamSettingsService upstreamSettingsService,
                                              MirroringValidationHelper validationHelper) {
        this.contentHashService = contentHashService;
        this.dedupingExecutor = new DedupingExecutor(executorService);
        this.eventPublisher = eventPublisher;
        this.executorService = executorService;
        this.projectMappingDao = projectMappingDao;
        this.projectService = projectService;
        this.repoMappingDao = repoMappingDao;
        this.repositoryService = repositoryService;
        this.scmService = scmService;
        this.transactionTemplate = transactionTemplate;
        this.upstreamClientFactory = upstreamClientFactory;
        this.upstreamSettingsService = upstreamSettingsService;
        this.validationHelper = validationHelper;

        maxFailedSyncCount = config.getMaxSyncFailedCount();
        repositoryFetchExecutor = concurrencyService.getBucketedExecutor(SYNC_EXECUTOR_NAME,
                new BucketedExecutorSettings.Builder<>(RepositoryFetchTask::getHierarchyId, repositoryFetchProcessor)
                        .batchSize(Integer.MAX_VALUE)
                        .maxAttempts(1)
                        .maxConcurrency(config.getMaxSyncThreadCount(), ConcurrencyPolicy.PER_CLUSTER)
                        .build());
    }

    @Override
    public void createLocalRepository(@Nonnull UpstreamServer upstream, @Nonnull MinimalExternalRepository externalRepository) {
        checkNotNull(upstream, "upstream");
        checkNotNull(externalRepository, "externalRepository");

        validationHelper.checkIsUpstreamUser(upstream.getId());

        inTransactionVoid(() -> {
            if (isMaybeMirrored(upstream, externalRepository)) {
                executorService.submit(new RepositorySynchronizationTask(upstream, externalRepository.getId(), empty(),
                        SyncLevel.DEFAULT));
            } else {
                log.trace("{}: Skipping synchronization of new repository because the repository is not mirrored",
                        describe(externalRepository));
            }
        });
    }

    @Override
    public int getMaxFailedSyncCount() {
        return maxFailedSyncCount;
    }

    @Override
    public void onStart() {
        // no-op
    }

    @Override
    public void onStop() {
        log.debug("Shutting down mirror synchronization executor");
        repositoryFetchExecutor.shutdown();
    }

    @Override
    public void startFullSynchronization(@Nonnull UpstreamServer upstream, @Nonnull SyncLevel syncLevel) {

        checkNotNull(upstream, "upstream");
        checkNotNull(syncLevel, "syncLevel");

        validationHelper.checkIsUpstreamUserOrAdmin(upstream.getId());

        dedupingExecutor.submit(upstream.getId(), new FullSynchronizationTask(upstream, syncLevel));
    }

    @Override
    public void startRepositorySynchronization(@Nonnull UpstreamServer upstream,
                                               @Nonnull MinimalExternalRepository externalRepository,
                                               @Nonnull List<RefChange> changes) {
        checkNotNull(upstream, "upstream");
        checkNotNull(externalRepository, "externalRepository");
        checkNotNull(changes, "changes");

        validationHelper.checkIsUpstreamUser(upstream.getId());

        inTransactionVoid(() -> {
            if (isMaybeMirrored(upstream, externalRepository)) {
                executorService.submit(new RepositorySynchronizationTask(upstream, externalRepository.getId(),
                        of(changes), SyncLevel.INCLUDE_FAILED));
            } else {
                log.trace("{}: Skipping synchronization because the repository is not mirrored",
                        describe(externalRepository));
            }
        });
    }

    @Override
    public void startRepositorySynchronization(@Nonnull UpstreamServer upstream, @Nonnull String externalRepositoryId,
                                               @Nonnull SyncLevel syncLevel) {
        checkNotNull(upstream, "upstream");
        checkNotNull(externalRepositoryId, "externalRepositoryId");
        checkNotNull(syncLevel, "syncLevel");

        validationHelper.checkIsUpstreamUserOrAdmin(upstream.getId());

        executorService.submit(new RepositorySynchronizationTask(upstream, externalRepositoryId, Optional.empty(),
                syncLevel));
    }

    @Override
    public void deleteLocalRepository(@Nonnull UpstreamServer upstream, @Nonnull MinimalExternalRepository externalRepository) {
        checkNotNull(upstream, "upstream");
        checkNotNull(externalRepository, "externalRepository");

        validationHelper.checkIsUpstreamUser(upstream.getId());

        deleteLocalRepository(upstream, externalRepository.getId(), true);
    }

    @Override
    public void updateLocalProject(@Nonnull UpstreamServer upstream, @Nonnull ExternalProject externalProject) {
        checkNotNull(upstream, "upstream");
        checkNotNull(externalProject, "externalProject");

        validationHelper.checkIsUpstreamUser(upstream.getId());

        syncLocalProject(upstream, externalProject, false);
    }

    @Override
    public void updateLocalRepository(@Nonnull UpstreamServer upstream, @Nonnull MinimalExternalRepository externalRepository) {
        checkNotNull(upstream, "upstream");
        checkNotNull(externalRepository, "externalRepository");

        validationHelper.checkIsUpstreamUser(upstream.getId());

        inTransactionVoid(() -> {
            if (isMaybeMirrored(upstream, externalRepository)) {
                executorService.submit(new RepositorySynchronizationTask(upstream, externalRepository.getId(), empty(),
                        SyncLevel.DEFAULT));
            } else {
                log.trace("{}: Skipping update because the repository is not mirrored", describe(externalRepository));
            }
        });
    }

    private boolean applyRefChangesLocally(Repository localRepository, List<RefChange> changes) {
        for (RefChange change : changes) {
            // if any of the RefChanges is incorrect, there's no point in calling updateRefs. We'll need a full sync
            if (StringUtils.isBlank(change.getFromHash()) || StringUtils.isBlank(change.getToHash()) ||
                    change.getRef() == null || StringUtils.isBlank(change.getRef().getId())) {
                return false;
            }
        }

        try {
            ScmMirrorCommandFactory commandFactory = scmService.getMirrorCommandFactory(localRepository);
            commandFactory.updateRefs(new MirrorUpdateRefsCommandParameters.Builder().changes(changes).build())
                    .call();
            return true;
        } catch (ServiceException e) {
            log.debug("{}: not all ref changes could be applied on the mirror ({})", localRepository, e.getMessage());
            return false;
        }
    }

    private Predicate<ExternalRepository> buildCanMirrorPredicate(UpstreamServer upstreamServer) {
        UpstreamSettings settings = getSettingsOrThrow(upstreamServer);

        return repo -> isMirrored(upstreamServer, settings, repo);
    }

    private boolean deleteLocalProjectIfEmpty(Project localProject) {
        if (!isEmpty(localProject)) {
            log.debug("{}: local project contains repositories so can't be deleted", localProject);
            return false;
        }

        log.debug("{}: deleting project mapping if it exists", localProject);
        projectMappingDao.delete(localProject.getId());
        log.info("{}: deleting this local project as it no longer contains mirrored repositories", localProject);
        projectService.delete(localProject);
        return true;
    }

    private void deleteLocalRepository(UpstreamServer upstream, String externalRepositoryId, boolean deleteLocalProjectIfEmpty) {
        inTransactionVoid(() -> {
            AoRepositoryMapping repositoryMapping = repoMappingDao.getByUpstreamId(upstream.getId(), externalRepositoryId);
            if (repositoryMapping != null) {
                Repository localRepository = repositoryService.getById(repositoryMapping.getLocalId());
                if (localRepository != null) {
                    Project localProject = localRepository.getProject();

                    log.info("{}: deleting this repository as its counterpart with id {} in upstream {} has been deleted",
                            localRepository, repositoryMapping.getExternalId(), describe(upstream));
                    repoMappingDao.delete(repositoryMapping.getUpstreamId(), repositoryMapping.getExternalId());
                    repositoryService.delete(localRepository);

                    if (deleteLocalProjectIfEmpty) {
                        deleteLocalProjectIfEmpty(localProject);
                    }
                } else {
                    log.info("Local mirror no longer exists for repository with id {} in upstream {} - cleaning up",
                            repositoryMapping.getExternalId(), describe(upstream));
                    repoMappingDao.delete(repositoryMapping.getUpstreamId(), repositoryMapping.getExternalId());
                }
            }
        });
    }

    private void failIfClashingLocalProjectFor(UpstreamServer upstream, ExternalProject externalProject, @Nullable Project existingLocalProject) {
        failIfClashingLocalProject(externalProject, projectService.getByKey(upstream.getNamespace(), externalProject.getKey()), "key", existingLocalProject);
        failIfClashingLocalProject(externalProject, projectService.getByName(upstream.getNamespace(), externalProject.getName()), "name", existingLocalProject);
    }

    private void failIfClashingLocalProject(ExternalProject externalProject, Project clashingLocalProject, String clashingProperty, @Nullable Project existingLocalProject) {
        // existingLocalProject will be non-null if the operation is a project rename / rekey (we don't want the existing project
        // to count as a clash)
        if (clashingLocalProject != null
                && (existingLocalProject == null || existingLocalProject.getId() != clashingLocalProject.getId())
                && !deleteLocalProjectIfEmpty(clashingLocalProject)) {

            throw new IncompatibleMirroredEntityStateException(
                    String.format("%s: synchronization failed - a local project (%s) with the same %s already exists but " +
                            "is unrelated to the upstream project", describe(externalProject), clashingLocalProject, clashingProperty));
        }
    }

    private void failIfClashingLocalRepoFor(UpstreamServer upstream, ExternalRepository externalRepository) {
        Repository localRepository = repositoryService.getBySlug(upstream.getNamespace(), externalRepository.getProject().getKey(), externalRepository.getSlug());
        if (localRepository != null) {
            throw new IncompatibleMirroredEntityStateException(
                    String.format("%s: synchronization failed - a local repository with the same slug (%s) already exists but " +
                            "is unrelated to the upstream repository", describe(externalRepository), localRepository));
        }
    }

    private UpstreamSettings getSettingsOrThrow(UpstreamServer upstreamServer) {
        return upstreamSettingsService.getSettings(upstreamServer.getId())
                .orElseThrow(() -> newNoUpstreamConfigurationException(upstreamServer));
    }

    private boolean isMaybeMirrored(@Nonnull UpstreamServer upstream, @Nonnull MinimalExternalRepository externalRepository) {
        UpstreamSettings settings = getSettingsOrThrow(upstream);

        if (!settings.isMirrored(externalRepository.getProject().getId())) {
            return false;
        }

        if (externalRepository instanceof ExternalRepository) {
            return isSupported((ExternalRepository) externalRepository);
        }

        // if we can't be certain if the repo should be mirrored because we don't have enough info then assume true
        // so that the caller can fetch and be certain
        return true;
    }

    private boolean isMirrored(UpstreamServer upstream, UpstreamSettings settings, ExternalRepository externalRepository) {
        if (!isSupported(externalRepository)) {
            log.trace("{}/{}/{}: skipping synchronization because there is no mirroring support for {} repositories",
                    upstream.getNamespace(), externalRepository.getProject().getKey(), externalRepository.getSlug(),
                    externalRepository.getScmId());
            return false;
        }

        if (!settings.isMirrored(externalRepository.getProject().getId())) {
            log.trace("{}/{}/{}: skipping synchronization because this repository is not mirrored",
                    upstream.getNamespace(), externalRepository.getProject().getKey(), externalRepository.getSlug());
            return false;
        }
        return true;
    }

    private IllegalStateException newNoUpstreamConfigurationException(UpstreamServer upstreamServer) {
        return new IllegalStateException("Upstream " + upstreamServer.getId() + " exists, but its configuration does not");
    }

    private String normalizeHttpCloneUrl(UpstreamServer upstream, String cloneUrl) {
        // Ensure the clone URL scheme always matches the base URL scheme - the correct thing to do but also helpful for
        // dev testing cloud where clone URLs are always "https" even if the base URL is not
        //
        // Although this normalization could only ever affect cloud upstreams in practice (server upstreams hand out
        // SSH clone URLs) we are being extra careful to ensure this is always the case
        if (upstream.getType() == UpstreamServerType.BITBUCKET_CLOUD) {
            String cloneUrlScheme = cloneUrl.split(":")[0];
            if (cloneUrlScheme.startsWith("http")) {
                String baseUrl = upstream.getBaseUrl();
                String baseUrlScheme = baseUrl.split(":")[0];
                if (!cloneUrlScheme.equals(baseUrlScheme)) {
                    return baseUrlScheme + cloneUrl.substring(cloneUrlScheme.length());
                }
            }
        }

        return cloneUrl;
    }

    private Project syncLocalProject(ExternalProject externalProject, SyncContext context) {
        try {
            return syncLocalProject(context.getUpstream(), externalProject, true);
        } catch (RuntimeException e) {
            log.error("{}/{}: failed to prepare local project: {}", context.getUpstream().getNamespace(),
                    externalProject.getKey(), e.getMessage());
            log.debug("Error details", e);
        }
        return null;
    }

    private Project syncLocalProject(UpstreamServer upstream, ExternalProject externalProject, boolean createIfNotExists) {
        return inTransaction(() -> {
            Project localProject;

            // no mapping means no local project and if we're not meant to create it then we have nothing more to do
            AoProjectMapping projectMapping = projectMappingDao.getByUpstreamId(upstream.getId(), externalProject.getId());
            if (projectMapping == null && !createIfNotExists) {
                return null;
            }

            if (projectMapping != null) {
                localProject = projectService.getById(projectMapping.getLocalId());
                if (localProject != null) {
                    // there's already a mapping and the project exists
                    return maybeUpdateProject(upstream, localProject, externalProject);
                }

                log.warn("A mapping between external project {} for upstream {} and local project {} exists but the local project " +
                                "no longer exists. Deleting the local mapping and reconstructing as necessary",
                        externalProject.getId(), describe(upstream), projectMapping.getLocalId());
                projectMappingDao.delete(projectMapping.getUpstreamId(), projectMapping.getExternalId());
            }

            // ensure there are no projects with the same key or name. this may happen if we've missed web hooks that
            // should have, for example, deleted a repo and project and now we're trying to rename or rekey to that project
            failIfClashingLocalProjectFor(upstream, externalProject, null);

            // project does not exist locally - create it
            //
            // note: local cloud projects will always be private due to a mismatch between server and cloud public permission
            // models. See RestCloudExternalProject#isPublic
            localProject = projectService.create(new ProjectCreateRequest.Builder()
                    .namespace(upstream.getNamespace())
                    .name(externalProject.getName())
                    .key(externalProject.getKey())
                    .publiclyAccessible(externalProject.isPublic())
                    .build());
            projectMappingDao.create(localProject.getId(), upstream.getId(), externalProject.getId());
            log.info("Created project '{}/{}' to mirror project from upstream server '{}'",
                    localProject.getNamespace(), localProject.getKey(), describe(upstream));
            return localProject;
        });
    }

    private Repository syncRepository(ExternalRepository externalRepository, SyncContext context) {
        try (Timer ignored = TimerUtils.start(TIMER_SYNC_LOCAL_REPOSITORY)) {
            UpstreamServer upstream = context.getUpstream();
            // #1 create or update the origin if present
            // if there's no origin, there's nothing to do
            // if there's no upstream account, all repositories can be mirrored
            Optional<UpstreamAccount> account = upstream.getAccount();
            Optional<Repository> origin = externalRepository.getOrigin()
                    .filter(orig -> !account.isPresent() || account.equals(orig.getOwner()))
                    .map(orig -> context.syncOnce(orig, this::syncRepository));

            // #2 create or update the target project
            Project targetProject = context.syncOnce(externalRepository.getProject(), this::syncLocalProject);
            if (targetProject == null) {
                return null;
            }

            // #3 sync repository meta data
            Repository repository = createOrUpdateLocalRepository(context.getUpstream(), externalRepository,
                    targetProject, origin.orElse(null));
            if (repository == null) {
                return null;
            }

            // #4 schedule fetches if out of sync
            AoRepositoryMapping mapping = inTransaction(() -> repoMappingDao.getByLocalId(repository.getId()));
            // If we get here, the database transactions that updated the project and/or repository have committed. At
            // this point it is safe to schedule the fetches
            getFetchTaskFor(context.getUpstream(), externalRepository, repository, mapping, context.getSyncLevel())
                    .ifPresent(context::fetch);
            return repository;
        } catch (RuntimeException e) {
            log.error("{}/{}/{}: failed to prepare local repository: {}", context.getUpstream().getNamespace(),
                    externalRepository.getProject().getKey(), externalRepository.getSlug(), e.getMessage());
            log.debug("Error details", e);
        }
        return null;
    }

    private Repository createOrUpdateLocalRepository(UpstreamServer upstream, ExternalRepository externalRepository,
                                                     Project targetProject, Repository targetOrigin) {
        return inTransaction(() -> {
            Repository localRepository = null;

            // check if there is a mapping already
            AoRepositoryMapping repoMapping = repoMappingDao.getByUpstreamId(
                    upstream.getId(), externalRepository.getId());

            if (repoMapping != null) {
                localRepository = repositoryService.getById(repoMapping.getLocalId());
                if (localRepository != null) {
                    return maybeUpdateRepository(upstream, targetProject, localRepository, externalRepository);
                }

                log.warn("A mapping between external repository {} for upstream {} and local repository {} exists but " +
                        "the local repository no longer exists. Deleting the local mapping and reconstructing as " +
                        "necessary",
                        externalRepository.getId(), describe(upstream), repoMapping.getLocalId());
                repoMappingDao.delete(repoMapping.getUpstreamId(), repoMapping.getExternalId());
            }

            // ensure there is no repo with the same slug. this may happen if we've missed web hooks that
            // should have deleted or renamed the repo out of the way
            failIfClashingLocalRepoFor(upstream, externalRepository);

            // create a new mapping and repository
            if (targetOrigin != null) {
                // set name to the upstream slug and thus the slug is the sluggified version of the upstream slug
                // to avoid validation issues since the slug cannot be set directly
                localRepository = repositoryService.fork(new RepositoryForkRequest.Builder(targetOrigin)
                        .project(targetProject)
                        .name(externalRepository.getSlug())
                        .publiclyAccessible(externalRepository.isPublic())
                        .build());
            } else {
                // set name to the upstream slug and thus the slug is the sluggified version of the upstream slug
                // to avoid validation issues since the slug cannot be set directly
                localRepository = repositoryService.create(new RepositoryCreateRequest.Builder()
                        .project(targetProject)
                        .scmId(externalRepository.getScmId())
                        .name(externalRepository.getSlug())
                        .publiclyAccessible(externalRepository.isPublic())
                        .build());
            }

            repoMappingDao.create(localRepository.getId(), upstream.getId(), localRepository.getProject().getId(),
                    externalRepository.getId());
            log.info("Created repository '{}' to mirror repository from upstream server '{}'",
                    localRepository, describe(upstream));

            return localRepository;
        });
    }

    private Optional<RepositoryFetchTask> getFetchTaskFor(UpstreamServer upstream,
                                                          ExternalRepository externalRepository,
                                                          Repository localRepository,
                                                          AoRepositoryMapping repositoryMapping,
                                                          SyncLevel syncLevel) {
        if (syncLevel == SyncLevel.DEFAULT && repositoryMapping.getFailedSyncCount() >= maxFailedSyncCount) {
            log.info("{}: skipping fetch because this repository has failed to fetch too many times", localRepository);
            return empty();
        }
        if (syncLevel != SyncLevel.FORCE_ALL && !hashesDiffer(localRepository, externalRepository)) {
            return empty();
        }

        Optional<String> cloneUrl = externalRepository.getCloneUrl();
        if (!cloneUrl.isPresent()) {
            log.warn("{}: skipping fetch because the clone URL cannot be determined", localRepository);
            return empty();
        }

        cloneUrl = cloneUrl.map(url -> normalizeHttpCloneUrl(upstream, url));

        return of(new RepositoryFetchTask(upstream.getId(), localRepository.getId(), localRepository.getHierarchyId(),
                externalRepository.getContentHash().get(), cloneUrl.get()));
    }

    private void getRepositoryUpstreamAndThen(UpstreamServer upstream, String externalRepositoryId,
                                              Consumer<ExternalRepository> syncOperation) {

        ExternalRepository externalRepository = upstreamClientFactory.create(upstream).getRepository(externalRepositoryId)
                .fail(t -> logUpstreamRequestFailure("Error retrieving repository {} from upstream {}",
                        externalRepositoryId, upstream.getBaseUrl(), t))
                .claim();

        inTransactionVoid(() -> syncOperation.accept(externalRepository));
    }

    private boolean hashesDiffer(Repository localRepository, ExternalRepository externalRepository) {
        Optional<String> externalHash = externalRepository.getContentHash();
        if (!externalHash.isPresent()) {
            log.warn("{}: skipping fetch because upstream content hash unavailable", localRepository);
            // If content hash retrieval failed on an upstream repository then we don't fetch because we don't know
            // the content hash to set locally and so otherwise the sync algorithm would re-fetch indefinitely
            return false;
        }

        Optional<String> localHash = contentHashService.getHash(localRepository);
        if (!localHash.isPresent()) {
            log.warn("{}: skipping fetch because local content hash unavailable", localRepository);
            // If local content hash retrieval failed then we don't fetch because we don't know whether the external
            // hash means our local repo is out of date and the sync algorithm would re-fetch indefinitely
            return false;
        }

        String external = externalHash.get();
        String local = localHash.get();
        if (local.equals(external)) {
            log.trace("{}: skipping fetch because upstream and local content hashes the same: {}", localRepository, external);
            return false;
        } else {
            log.debug("{}: upstream and local content hashes differ: {} <-> {}",
                    localRepository, external, local);
            return true;
        }
    }

    private <T> T inTransaction(TransactionCallback<T> t) {
        return transactionTemplate.execute(t);
    }

    private Void inTransactionVoid(Runnable r) {
        return inTransaction(() -> {
            r.run();
            return null;
        });
    }

    private boolean isEmpty(Project project) {
        return repositoryService.findByProjectKey(project.getNamespace(), project.getKey(), PageUtils.newRequest(0, 1))
                .getSize() == 0;
    }

    private boolean isSupported(ExternalRepository externalRepository) {
        try {
            return scmService.isSupported(externalRepository.getScmId(), ScmFeature.MIRRORS);
        } catch (UnavailableScmException | UnsupportedScmException e) {
            return false;
        }
    }

    private void logUpstreamRequestFailure(String message, Object... arguments) {
        if (arguments.length > 0 && arguments[arguments.length - 1] instanceof UnauthorizedException) {
            // When the mirror has been disabled in the upstream, some upstream requests will get an 401 response.
            // Since this is expected, this should not be WARN logged
            log.debug(message, arguments);
        } else {
            log.warn(message, arguments);
        }
    }

    private Project maybeUpdateProject(UpstreamServer upstream, Project project, ExternalProject externalProject) {
        ProjectUpdateRequest.Builder builder = new ProjectUpdateRequest.Builder(project);
        boolean needsUpdate = false;

        boolean isRekeyed = !project.getKey().equalsIgnoreCase(externalProject.getKey());
        boolean isRenamed = !project.getName().equals(externalProject.getName());

        if (isRekeyed || isRenamed) {
            failIfClashingLocalProjectFor(upstream, externalProject, project);
        }

        if (isRekeyed) {
            log.info("{}: rekeying to {}", project, externalProject.getKey());
            builder = builder.key(externalProject.getKey());
            needsUpdate = true;
        }

        if (isRenamed) {
            log.info("{}: renaming to {}", project, externalProject.getName());
            builder = builder.name(externalProject.getName());
            needsUpdate = true;
        }

        // note: local cloud projects will always be private due to a mismatch between server and cloud public permission
        // models. See RestCloudExternalProject#isPublic
        if (externalProject.isPublic() != project.isPublic()) {
            log.info("{}: changing public flag to {}", project, externalProject.isPublic());
            builder.publiclyAccessible(externalProject.isPublic());
            needsUpdate = true;
        }

        if (needsUpdate) {
            return projectService.update(builder.build());
        }

        return project;
    }

    /**
     * Renames/reslugs and/or moves a local repository based on its external counterpart's current state. As a by-product of this
     * process the containing project may also be rekeyed if it, too, differs from its upstream counterpart.
     * <p>
     * Note that we aren't explicitly handling the case where a repo's origins is deleted. This will be handled through a webhook.
     */
    private Repository maybeUpdateRepository(UpstreamServer upstream, Project targetProject,
                                             Repository repository, ExternalRepository externalRepository) {
        RepositoryUpdateRequest.Builder updateRequest = new RepositoryUpdateRequest.Builder(repository);
        boolean needsUpdate = false;

        Project originalProject = repository.getProject();
        String originalSlug = repository.getSlug();

        String targetSlug = externalRepository.getSlug();

        boolean didMove = originalProject.getId() != targetProject.getId();
        boolean didReslug = !originalSlug.equalsIgnoreCase(targetSlug);

        if (didMove || didReslug) {
            failIfClashingLocalRepoFor(upstream, externalRepository);
        }

        if (didMove) {
            // set name to the updated upstream slug and thus the slug is the sluggified version of the upstream slug
            // to avoid validation issues since the slug cannot be set directly
            log.info("{}: moving repository to {}", repository, targetProject);
            updateRequest = updateRequest.project(targetProject);
            needsUpdate = true;
        }

        if (didReslug) {
            log.info("{}: renaming to {}", repository, targetSlug);
            updateRequest = updateRequest.name(targetSlug);
            needsUpdate = true;
        }

        if (repository.isPublic() != externalRepository.isPublic()) {
            log.info("{}: changing public flag to {}", repository, externalRepository.isPublic());
            updateRequest.publiclyAccessible(externalRepository.isPublic());
            needsUpdate = true;
        }

        if (needsUpdate) {
            Repository update = repositoryService.update(updateRequest.build());
            if (didMove) {
                repoMappingDao.updateProject(repository.getId(), targetProject.getId());
                deleteLocalProjectIfEmpty(originalProject);
            }
            return update;
        }

        return repository;
    }


    private void syncLocalRepositoryAndFetch(UpstreamServer upstream, ExternalRepository externalRepository,
                                             Optional<List<RefChange>> changes, SyncLevel syncLevel) {

        SyncContext context = new SyncContext(upstream, syncLevel);
        Repository localRepository = context.syncOnce(externalRepository, this::syncRepository);

        // if ref changes are available and the list is non-empty then try to apply the ref updates first
        // (they might be unavailable if this method is called as part of a repo create or update action)
        boolean changesApplied = localRepository != null && changes.filter(refChanges -> !refChanges.isEmpty())
                .map(refChanges -> {
                    if (applyRefChangesLocally(localRepository, refChanges)) {
                        log.debug("{}: all upstream ref changes applied cleanly. No fetch is required.", localRepository);

                        if (externalRepository.getContentHash().isPresent()) {
                            contentHashService.setHash(localRepository, externalRepository.getContentHash().get());
                        }
                        eventPublisher.publish(new RepositorySynchronizedEvent(
                                this, localRepository, refChanges, Collections.emptySet(), upstream.getId()));

                        return true;
                    } else {
                        log.debug("{}: could not apply all ref changes locally.", localRepository);
                    }
                    return false;
        }).orElse(false);

        context.getFetchTasks().forEach(task -> {
            // find the task, if it exists, that corresponds to the repo for which the original sync has been requested
            if (localRepository.getId() == task.getRepositoryId()) {
                if (changesApplied) {
                    // changes have already been applied - no need to fetch
                    return;
                }

                // if changes are available and the list is non-empty and we're dealing with a server upstream then we
                // can skip the fetch because the change is from a remotely-merged PR event or a ref sync where all refs
                // were rejected so there were no ref changes
                //
                // for cloud a provided but empty change list means a non-standard ref changed and so we should fetch
                if (changes.isPresent() && changes.get().isEmpty() && upstream.getType() == BITBUCKET_SERVER) {
                    log.trace("{}: No ref changes published. Skipping repository fetch", describe(externalRepository));
                    return;
                }
            }

            // fallthrough: either this is not the target repository but a repository in the origin chain, the changes
            // didn't fully apply, there were no changes (a repository sync for another reason like a repo-created
            // web hook) or in the case of cloud upstreams non-standard refs may have been changed
            log.debug("{}: Fetch scheduled", localRepository);
            repositoryFetchExecutor.submit(task);
        });
    }

    /**
     * Synchronises the state of a local repository with its upstream counterpart after making a remote call to the upstream
     * to determine its upstream counterpart's state.
     */
    private void syncRepository(UpstreamServer upstream, String externalRepositoryId,
                                Predicate<ExternalRepository> canMirror, Optional<List<RefChange>> changes,
                                SyncLevel syncLevel, boolean deleteLocalProjectIfEmpty) {

        getRepositoryUpstreamAndThen(upstream, externalRepositoryId, externalRepository -> {
            if (externalRepository == null || !canMirror.test(externalRepository)) {
                deleteLocalRepository(upstream, externalRepositoryId, deleteLocalProjectIfEmpty);
            } else {
                syncLocalRepositoryAndFetch(upstream, externalRepository, changes, syncLevel);
            }
        });
    }

    /**
     * @return a stream of ExternalRepositories starting with the supplied repository working its way up the origin
     * hierarchy
     */
    private static Stream<ExternalRepository> toHierarchyStream(ExternalRepository repo) {
        return Stream.concat(Stream.of(repo),
                repo.getOrigin().map(DefaultMirrorSynchronizationHelper::toHierarchyStream).orElse(Stream.empty()));
    }

    abstract class BaseSynchronizeRepositoriesTask implements Runnable {

        protected final InternalUpstreamClient client;
        protected final SyncLevel syncLevel;
        protected final UpstreamServer upstream;

        public BaseSynchronizeRepositoriesTask(SyncLevel syncLevel, UpstreamServer upstream) {
            this.client = upstreamClientFactory.create(checkNotNull(upstream, "upstream"));
            this.syncLevel = checkNotNull(syncLevel, "syncLevel");
            this.upstream = checkNotNull(upstream, "upstream");
        }

        protected void syncSingleRepository(String externalRepositoryId, Optional<List<RefChange>> changes,
                                            SyncLevel syncLevel, boolean deleteLocalProjectIfEmpty) {
            try {
                log.debug("{}/{}: retrieving repository from upstream server", describe(upstream), externalRepositoryId);
                // a per-repository sync is indicative of upstream activity so even if we previously marked
                // it as failed we should still sync in case the the activity was to repair it
                syncRepository(upstream, externalRepositoryId, buildCanMirrorPredicate(upstream), changes,
                        syncLevel, deleteLocalProjectIfEmpty);
            } catch (UnauthorizedException e) {
                log.info("{}/{}: repository synchronization failed because the upstream rejected the mirror's " +
                                "credentials. Perhaps the mirror is disabled in the upstream?",
                        describe(upstream), externalRepositoryId);
            } catch (RuntimeException e) {
                log.error("{}/{} repository synchronization failed", describe(upstream), externalRepositoryId, e);
            }
        }
    }

    /**
     * Retrieves the repositories that should be mirrored from the upstream server and schedules a repository sync for
     * each repository that is out of date.
     */
    class FullSynchronizationTask extends BaseSynchronizeRepositoriesTask {

        private FullSynchronizationTask(UpstreamServer upstream, SyncLevel syncLevel) {
            super(syncLevel, upstream);
        }

        @Override
        public void run() {
            String upstreamDescription = describe(upstream);

            try (Timer ignored = TimerUtils.start(TIMER_FULL_SYNCHRONIZATION)) {
                Date startDate = new Date();

                Predicate<ExternalRepository> canMirror = buildCanMirrorPredicate(upstream);

                log.debug("{}: Synchronizing upstream repositories", upstreamDescription);
                FullSyncResult fullSyncResult = syncAllExternalRepositories(canMirror);
                log.debug("{}: Finished synchronizing upstream repositories", upstreamDescription);

                // there three main anticipated reasons why the full sync may not have completed successfully:
                // 1. one or more repos were deleted upstream thus we didn't see them in any of the pages
                // 2. due to paging misalignment caused by live updates to the mirrored repo list we
                // skipped one or more repos when paging through
                // 3. a batch of repo/project updates failed and was rolled back
                // in all three cases we fix this by doing individual syncs for the upstream repo IDs concerned
                // in the case of a deleted repo, we end up deleting it locally.
                // in the case of a skipped repo, we do a sync and possibly a fetch.
                // in the case of repos in a failing batch, we do a sync and possibly a fetch and, only if there are
                // still problems, will the repo in error fail to sync - others in the batch should complete without error.
                // note: we sync deleted/missing repos first because by deleting them and their potentially empty
                // projects we remove any constraint violations that might have caused repos in batches to fail
                // (e.g. creation or rename of repos/projects with same key/slug/name as a deleted repo/project) thus
                // improving the changes of all repos to sync successfully

                if (!fullSyncResult.isComplete()) {
                    Set<String> unseenExternalRepoIds = fullSyncResult.getUnseenExternalRepoIds();
                    Set<String> unsyncedExternalRepoIds = fullSyncResult.getUnsyncedExternalRepoIds();

                    // order is important - process deleted/unseen repo ids first then ones that failed to sync
                    for (Set<String> externalRepoIds : asList(unseenExternalRepoIds, unsyncedExternalRepoIds)) {
                        if (!externalRepoIds.isEmpty()) {
                            // run individual syncs for deleted or missing repos
                            syncRepositoriesIndividually(upstreamDescription, externalRepoIds);

                            // clean up empty projects from any deleted repos
                            inTransactionVoid(this::deleteEmptyLocalProjectsAndMappings);
                        }
                    }
                }

                eventPublisher.publish(new FullSynchronizationEvent(this, startDate, upstream));
            } catch (UpstreamServerRequestRateExceededException e) {
                log.warn("{}: full synchronization failed because an upstream request rate limit has been exceeded. " +
                        "Aborting full sync", upstreamDescription);
            } catch (UnauthorizedException e) {
                log.info("{}: full synchronization failed because the upstream rejected the mirror's credentials. " +
                        "Perhaps the mirror is disabled in the upstream?", upstreamDescription);
            } catch (RuntimeException e) {
                log.error("{}: full synchronization failed", upstreamDescription, e);
            }
        }

        private void deleteEmptyLocalProjectsAndMappings() {
            Set<Integer> allLocalProjectIds = projectMappingDao.getLocalIdsByUpstreamId(upstream.getId());
            if (allLocalProjectIds.isEmpty()) {
                return;
            }

            Set<Integer> nonEmptyLocalProjectIds = repoMappingDao.getLocalProjectIdsByUpstreamId(upstream.getId());
            Set<Integer> emptyLocalProjectIds = Sets.difference(allLocalProjectIds, nonEmptyLocalProjectIds);

            for (Integer emptyLocalProjectId : emptyLocalProjectIds) {
                Project emptyLocalProject = projectService.getById(emptyLocalProjectId);
                if (emptyLocalProject == null) {
                    log.warn("A project mapping refers to a local project with id {} but this no longer exists");
                    continue;
                }

                if (!isEmpty(emptyLocalProject)) {
                    log.warn("{}: one or more repositories exist in this local project but no repository mappings refer to it. It cannot be deleted",
                            emptyLocalProject);
                    continue;
                }

                log.debug("{}: deleting project mapping if it exists", emptyLocalProject);
                projectMappingDao.delete(emptyLocalProject.getId());
                log.info("{}: deleting local project as it no longer contains mirrored repositories", emptyLocalProject);
                projectService.delete(emptyLocalProject);
            }
        }

        private Set<String> findAllLocallyKnownExternalRepoIds() {
            Iterable<AoRepositoryMapping> repositoryMappings =
                    new PagedIterable<>((PageRequest pageRequest) ->
                            repoMappingDao.findByUpstreamId(upstream.getId(), pageRequest), REPO_MAPPING_PAGE_MAX);

            return inTransaction(() ->
                    streamIterable(repositoryMappings)
                            .map(AoRepositoryMapping::getExternalId)
                            .collect(toSet()));
        }

        private FullSyncResult syncAllExternalRepositories(Predicate<ExternalRepository> canMirror) {
            Set<String> allExternalRepoIdsSeen = new HashSet<>();
            Set<String> syncedExternalRepoIds = new HashSet<>();

            // iterate over every upstream repo and after a batch of repositoryDbBatchSize repos are seen,
            // sync the batch
            SyncContext syncContext = new SyncContext(upstream, syncLevel);
            try {
                client.getRepositories().filter(canMirror).forEach(repository -> {
                    // add repo's id and origin's id etc to the set we've seen
                    toHierarchyStream(repository)
                            .map(ExternalRepository::getId)
                            .forEach(allExternalRepoIdsSeen::add);

                    if (syncContext.syncOnce(repository, DefaultMirrorSynchronizationHelper.this::syncRepository) != null) {
                        syncedExternalRepoIds.add(repository.getId());
                    }
                });
            } catch (UpstreamServerRequestRateExceededException | UnauthorizedException e) {
                // abruptly terminate full sync and let caller handle the error
                throw e;
            } catch (RuntimeException e) {
                // a network call has failed - there's not much we can do except perform any fetch task scheduling
                // required thus far and let the caller handle the error
                syncContext.getFetchTasks().forEach(repositoryFetchExecutor::submit);
                throw e;
            }

            syncContext.getFetchTasks().forEach(repositoryFetchExecutor::submit);
            return new FullSyncResult(allExternalRepoIdsSeen, syncedExternalRepoIds, findAllLocallyKnownExternalRepoIds());
        }

        private void syncRepositoriesIndividually(String upstreamDescription, Set<String> externalRepositoryIds) {
            log.debug("{}: Performing {} individual synchronizations for these repositories: {}",
                    upstreamDescription, externalRepositoryIds.size(),
                    externalRepositoryIds.stream().collect(Collectors.joining(", ")));

            for (String externalRepositoryId : externalRepositoryIds) {
                syncSingleRepository(externalRepositoryId, empty(), syncLevel, false);
            }

            log.debug("{}: Finished {} individual synchronizations for these repositories: {}",
                    upstreamDescription, externalRepositoryIds.size(),
                    externalRepositoryIds.stream().collect(Collectors.joining(", ")));
        }

        private class FullSyncResult {
            private final Set<String> unseenExternalRepoIds;
            private final Set<String> unsyncedExternalRepoIds;

            public FullSyncResult(Set<String> allExternalRepoIds, Set<String> syncedExternalRepoIds, Set<String> localExternalRepoIds) {
                // add any repo id which we didn't see from the upstream but we previously synced (so it's probably
                // deleted)
                unseenExternalRepoIds = Sets.difference(localExternalRepoIds, allExternalRepoIds);
                // add any repo id which we saw from the upstream but failed to sync for some reason
                unsyncedExternalRepoIds = Sets.difference(allExternalRepoIds, syncedExternalRepoIds);
            }

            public Set<String> getUnseenExternalRepoIds() {
                return unseenExternalRepoIds;
            }

            public Set<String> getUnsyncedExternalRepoIds() {
                return unsyncedExternalRepoIds;
            }

            public boolean isComplete() {
                return unsyncedExternalRepoIds.isEmpty() && unseenExternalRepoIds.isEmpty();
            }
        }
    }

    class RepositorySynchronizationTask extends BaseSynchronizeRepositoriesTask {
        /**
         * The optional ref changes that caused the task to be created and executed. An {@link Optional#empty() absent}
         * list means that ref changes did not prompt the task's creation. A present list means ref changes did cause
         * the task creation although the list may itself be empty since various cloud and server conditions may cause
         * web hook events to be fired with an empty change list.
         */
        private final Optional<List<RefChange>> changes;
        private final String externalRepositoryId;

        public RepositorySynchronizationTask(UpstreamServer upstream, String externalRepositoryId,
                                             Optional<List<RefChange>> changes, SyncLevel syncLevel) {
            super(syncLevel, upstream);

            this.changes = checkNotNull(changes, "changes");
            this.externalRepositoryId = checkNotNull(externalRepositoryId, "externalRepositoryId");
        }

        @Override
        public void run() {
            try (Timer ignored = TimerUtils.start(TIMER_REPOSITORY_SYNCHRONIZATION)) {
                syncSingleRepository(externalRepositoryId, changes, syncLevel, true);
            } catch (UpstreamServerRequestRateExceededException e) {
                log.warn("{}: synchronization of repository with external id {} failed because an " +
                        "upstream request rate limit has been exceeded",
                        describe(upstream), externalRepositoryId, e);
            }
        }
    }
}
