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

import com.atlassian.bitbucket.AuthorisationException;
import com.atlassian.bitbucket.Product;
import com.atlassian.bitbucket.auth.AuthenticationContext;
import com.atlassian.bitbucket.i18n.I18nService;
import com.atlassian.bitbucket.i18n.KeyedMessage;
import com.atlassian.bitbucket.internal.mirroring.SimpleMirroringRequest;
import com.atlassian.bitbucket.internal.mirroring.mirror.client.InternalUpstreamClientFactory;
import com.atlassian.bitbucket.internal.mirroring.mirror.dao.*;
import com.atlassian.bitbucket.internal.mirroring.mirror.feature.CloudMirroringFeature;
import com.atlassian.bitbucket.internal.mirroring.mirror.nav.MirroringNavBuilder;
import com.atlassian.bitbucket.internal.mirroring.mirror.sync.MirrorSynchronizationHelper;
import com.atlassian.bitbucket.mirroring.mirror.*;
import com.atlassian.bitbucket.permission.Permission;
import com.atlassian.bitbucket.permission.PermissionValidationService;
import com.atlassian.bitbucket.project.Project;
import com.atlassian.bitbucket.repository.RefChange;
import com.atlassian.bitbucket.server.ApplicationMode;
import com.atlassian.bitbucket.server.ApplicationPropertiesService;
import com.atlassian.bitbucket.server.FeatureManager;
import com.atlassian.bitbucket.server.IncompatibleApplicationModeException;
import com.atlassian.bitbucket.user.EscalatedSecurityContext;
import com.atlassian.bitbucket.validation.ArgumentValidationException;
import com.atlassian.event.api.EventListener;
import com.atlassian.event.api.EventPublisher;
import com.atlassian.johnson.event.EventLevel;
import com.atlassian.jwt.JwtIssuer;
import com.atlassian.jwt.JwtIssuerRegistry;
import com.atlassian.sal.api.transaction.TransactionCallback;
import com.atlassian.sal.api.transaction.TransactionTemplate;
import com.atlassian.util.concurrent.Promise;
import com.atlassian.util.concurrent.Promises;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableMap;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.net.MalformedURLException;
import java.net.URI;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.text.Normalizer;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import static com.atlassian.bitbucket.internal.mirroring.MirroringConstants.*;
import static com.atlassian.bitbucket.internal.mirroring.mirror.MirrorDescriptionUtils.describe;
import static com.atlassian.bitbucket.mirroring.mirror.IntegrationState.*;
import static com.atlassian.bitbucket.mirroring.mirror.UpstreamServerType.BITBUCKET_CLOUD;
import static com.atlassian.bitbucket.mirroring.mirror.UpstreamServerType.BITBUCKET_SERVER;
import static com.atlassian.util.concurrent.Promises.promise;
import static java.util.Objects.requireNonNull;
import static org.apache.commons.lang.StringUtils.abbreviate;

public class DefaultUpstreamService implements InternalUpstreamService, JwtIssuerRegistry {

    private static final long REFRESH_DELAY_MS = TimeUnit.SECONDS.toMillis(10);

    private static final Void VOID_RESULT = null;
    private static final Logger log = LoggerFactory.getLogger(DefaultUpstreamService.class);

    private final AuthenticationContext authenticationContext;
    private final MirroringConfig config;
    private final EventPublisher eventPublisher;
    private final ScheduledExecutorService executorService;
    private final FeatureManager featureManager;
    private final I18nService i18nService;
    private final JohnsonHelper johnsonHelper;
    private final MirroringNavBuilder mirroringNavBuilder;
    private final UpstreamUserHelper securityHelper;
    private final MirrorSynchronizationHelper mirrorSynchronizationHelper;
    private final PermissionValidationService permissionValidationService;
    private final ProjectMappingDao projectMappingDao;
    private final ApplicationPropertiesService propertiesService;
    private final RepositoryMappingDao repositoryMappingDao;
    private final TransactionTemplate transactionTemplate;
    private final InternalUpstreamClientFactory upstreamClientFactory;
    private final UpstreamServerDao upstreamServerDao;

    private boolean johnsoned;

    public DefaultUpstreamService(AuthenticationContext authenticationContext, MirroringConfig config,
                                  EventPublisher eventPublisher, ScheduledExecutorService executorService,
                                  FeatureManager featureManager, I18nService i18nService, JohnsonHelper johnsonHelper,
                                  MirroringNavBuilder mirroringNavBuilder,
                                  MirrorSynchronizationHelper mirrorSynchronizationHelper,
                                  PermissionValidationService permissionValidationService,
                                  ProjectMappingDao projectMappingDao, ApplicationPropertiesService propertiesService,
                                  RepositoryMappingDao repositoryMappingDao, UpstreamUserHelper securityHelper,
                                  TransactionTemplate transactionTemplate,
                                  InternalUpstreamClientFactory upstreamClientFactory,
                                  UpstreamServerDao upstreamServerDao) {

        this.authenticationContext = authenticationContext;
        this.config = config;
        this.eventPublisher = eventPublisher;
        this.executorService = executorService;
        this.featureManager = featureManager;
        this.i18nService = i18nService;
        this.johnsonHelper = johnsonHelper;
        this.mirroringNavBuilder = mirroringNavBuilder;
        this.mirrorSynchronizationHelper = mirrorSynchronizationHelper;
        this.permissionValidationService = permissionValidationService;
        this.projectMappingDao = projectMappingDao;
        this.propertiesService = propertiesService;
        this.repositoryMappingDao = repositoryMappingDao;
        this.securityHelper = securityHelper;
        this.transactionTemplate = transactionTemplate;
        this.upstreamClientFactory = upstreamClientFactory;
        this.upstreamServerDao = upstreamServerDao;
    }

    @Override
    public JwtIssuer getIssuer(@Nonnull String issuer) {
        requireNonNull(issuer, "issuer");

        InternalUpstreamServer upstream = inTransaction(() -> upstreamServerDao.getByIssuerId(issuer));
        return upstream == null ? null : new UpstreamServerJwtIssuer(upstream);
    }

    @Override
    public InternalUpstreamServer get() {
        if (!isMirror()) {
            return null;
        }

        return inTransaction(upstreamServerDao::getUpstream);
    }

    @Nonnull
    @Override
    public Optional<InternalUpstreamServer> getById(@Nonnull String upstreamId) {
        requireNonNull(upstreamId, "upstreamId");

        InternalUpstreamServer internalUpstreamServer = get();
        if (internalUpstreamServer == null || !internalUpstreamServer.getId().equals(upstreamId)) {
            return Optional.empty();
        } else {
            return Optional.of(internalUpstreamServer);
        }
    }

    @Nonnull
    @Override
    public InternalUpstreamServer createForServer(@Nonnull String url) throws IllegalStateException {
        requireNonNull(url, "url");

        if (!isValidUrl(url)) {
            throw new IllegalArgumentException(String.format("Upstream '%s' is not a valid URL.", url));
        }

        return createSingleUpstream(url, BITBUCKET_SERVER);
    }

    @Nonnull
    @Override
    public InternalUpstreamServer createForCloud() throws IllegalStateException {
        String upstreamUrl = getCloudUpstreamUrl().orElseThrow(() -> new IllegalStateException("The cloud URL is not configured."));
        return createSingleUpstream(upstreamUrl, BITBUCKET_CLOUD);
    }

    /**
     * @param upstream The upstream server
     * @return the synchronization progress state
     */
    @Nonnull
    @Override
    public DefaultSyncProgress getSynchronizationProgress(@Nonnull UpstreamServer upstream) {
        return inTransaction(() -> {
            int syncedRepos = repositoryMappingDao.countSyncedInProgress(upstream.getId());
            int totalRepos = repositoryMappingDao.countAllInProgress(upstream.getId());
            Date lastSettingsChangeDate = upstream.getLastSettingsChangeDate();
            Date lastFullSyncDate = upstream.getLastFullSyncDate();
            boolean discovering = lastSettingsChangeDate != null && (lastFullSyncDate == null ||
                    lastFullSyncDate.before(lastSettingsChangeDate));
            if (syncedRepos >= totalRepos && !discovering) {
                // Future calls to this method will return 0/0 repos
                projectMappingDao.clearAllProgress(upstream.getId());
            }
            return new DefaultSyncProgress(discovering, syncedRepos, totalRepos);
        });
    }

    @Override
    public boolean isMirror() {
        return propertiesService.getMode() == ApplicationMode.MIRROR;
    }

    @EventListener
    public void onFullSynchronizationEvent(FullSynchronizationEvent event) {
        inTransactionVoid(() -> upstreamServerDao.setLastFullSyncDate(event.getUpstream().getId(), event.getStartDate()));
    }

    @EventListener
    public void onRepositorySynchronizationFailed(RepositorySynchronizationFailedEvent event) {
        int failedSyncCount = inTransaction(() ->
                repositoryMappingDao.incrementFailedSyncCount(event.getRepository().getId()));
        if (failedSyncCount >= config.getMaxSyncFailedCount()) {
            log.warn("{}: fetch from upstream failed {} times in a row. Further fetches will not be attempted until the repository is next updated.",
                    event.getRepository(), failedSyncCount);
        }
        maybeClearProgress(event.getUpstreamServer());
    }

    @EventListener
    public void onRepositorySynchronized(RepositorySynchronizedEvent event) {
        inTransactionVoid(() -> repositoryMappingDao.updateSyncDate(event.getRepository().getId(), event.getDate()));
        maybeClearProgress(event.getUpstreamServer());
    }

    @EventListener
    public void onUpstreamSettingsChanged(UpstreamSettingsChangedEvent event) {
        if (settingsChangeNeedsFullSync(event)) {
            startFullSynchronizationAsUpstreamUser(event.getUpstream(), SyncLevel.DEFAULT);
        }
    }

    @Nonnull
    @Override
    public UpstreamServer onInstalled(@Nonnull UpstreamInstallationParameters installedParameters) {
        requireNonNull(installedParameters, "installedParameters");

        return inTransaction(() -> {
            String upstreamId = installedParameters.getUpstreamId();
            AoUpstreamServer upstream = getUpstreamOrFail(upstreamId);

            if (!isAuthenticatedAsUpstream(upstreamId)) {
                IntegrationState state = upstream.getState();
                if (state != INITIALIZING && state != PENDING) {
                    throw throwNotPermitted();
                }
            }

            if (upstream.getType() == BITBUCKET_CLOUD) {
                if (!installedParameters.getTeamId().isPresent()
                    || !installedParameters.getTeamName().isPresent()) {
                    throw throwBadRequest("bitbucket.mirroring.upstream.cloud.team.required");
                } else if (!installedParameters.getApiBaseUrl().isPresent()) {
                    throw throwBadRequest("bitbucket.mirroring.upstream.cloud.api.required");
                }
            }

            upstream = upstreamServerDao.setInstalledParameters(installedParameters);
            if (upstream == null) {
                throw throwNoSuchUpstream(upstreamId);
            }

            return setUpstreamServerIntegrationState(upstream, INSTALLED);
        });
    }

    @Override
    public void onProjectModified(@Nonnull String upstreamId, @Nonnull ExternalProject externalProject) {
        requireNonNull(upstreamId, "upstreamId");
        requireNonNull(externalProject, "externalProject");

        validateAuthenticatedAsUpstream(upstreamId);

        mirrorSynchronizationHelper.updateLocalProject(getUpstreamOrFail(upstreamId), externalProject);
    }

    @Override
    public void onRepositoryContentsChanged(@Nonnull String upstreamId,
                                            @Nonnull MinimalExternalRepository externalRepository,
                                            @Nonnull List<RefChange> changes) {
        requireNonNull(upstreamId, "upstreamId");
        requireNonNull(externalRepository, "externalRepository");
        requireNonNull(changes, "changes");

        validateAuthenticatedAsUpstream(upstreamId);

        mirrorSynchronizationHelper.startRepositorySynchronization(getUpstreamOrFail(upstreamId), externalRepository, changes);
    }

    @Override
    public void onRepositoryCreated(@Nonnull String upstreamId, @Nonnull MinimalExternalRepository externalRepository) {
        requireNonNull(upstreamId, "upstreamId");
        requireNonNull(externalRepository, "externalRepository");

        validateAuthenticatedAsUpstream(upstreamId);

        mirrorSynchronizationHelper.createLocalRepository(getUpstreamOrFail(upstreamId), externalRepository);
    }

    @Override
    public void onRepositoryDeleted(@Nonnull String upstreamId, @Nonnull MinimalExternalRepository externalRepository) {
        requireNonNull(upstreamId, "upstreamId");
        requireNonNull(externalRepository, "externalRepository");

        validateAuthenticatedAsUpstream(upstreamId);

        mirrorSynchronizationHelper.deleteLocalRepository(getUpstreamOrFail(upstreamId), externalRepository);
    }

    @Override
    public void onRepositoryModified(@Nonnull String upstreamId, @Nonnull MinimalExternalRepository externalRepository) {
        requireNonNull(upstreamId, "upstreamId");
        requireNonNull(externalRepository, "externalRepository");

        validateAuthenticatedAsUpstream(upstreamId);

        mirrorSynchronizationHelper.updateLocalRepository(getUpstreamOrFail(upstreamId), externalRepository);
    }

    @Override
    public void onUninstalled(@Nonnull String upstreamId) {
        requireNonNull(upstreamId, "upstreamId");

        onStateUpdated(upstreamId, REMOVED);
    }

    @Override
    public Promise<UpstreamServer> refresh(@Nonnull UpstreamServer upstream) {
        return refreshFromUpstreamNow(requireNonNull(upstream, "upstream"));
    }

    @Override
    public Promise<Void> register() {
        checkIsMirror();

        InternalUpstreamServer upstream = getOrCreateUpstreamServer();
        if (upstream.getState() == REMOVED) {
            // allow re-registering from the REMOVED state as well
            upstream = setUpstreamServerIntegrationState(upstream.getId(), initialStateFor(upstream));
        }

        //Cloud doesn't have a registration step.
        if (upstream.getType() != BITBUCKET_CLOUD
                && (upstream.getState() == INITIALIZING || upstream.getState() == PENDING)) {
            return registerUpstream(upstream);
        }

        return promise(VOID_RESULT);
    }

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

        permissionValidationService.validateForGlobal(Permission.ADMIN);

        refreshFromUpstreamNow(upstream)
                .done(updatedUpstream -> {
                    if (updatedUpstream.getState() == INSTALLED) {
                        mirrorSynchronizationHelper.startRepositorySynchronization(upstream, externalRepositoryId,
                                syncLevel);
                    } else {
                        log.debug("{}: skipping synchronization of {} because the mirror's state on the upstream server is '{}'",
                                describe(upstream), externalRepositoryId, updatedUpstream.getState());
                    }
                });
    }

    @Override
    public void startSynchronization(@Nonnull SyncLevel syncLevel) {
        requireNonNull(syncLevel, "syncLevel");

        if (!isMirror()) {
            throw notAMirrorException();
        }

        permissionValidationService.validateForGlobal(Permission.ADMIN);

        UpstreamServer upstream = get();
        if (upstream == null) {
            log.debug("Not synchronizing with upstream because this mirror has not been registered with one yet");
            throw notYetRegisteredException();
        }

        refreshFromUpstreamNow(upstream)
                .done(updatedUpstream -> {
                    if (updatedUpstream.getState() == INSTALLED) {
                        startFullSynchronizationAsUpstreamUser(updatedUpstream, syncLevel);
                    } else {
                        log.debug("{}: skipping full synchronization because the mirror's state on the upstream server is '{}'",
                                describe(upstream), updatedUpstream.getState());
                    }
                });
    }

    @Nonnull
    @Override
    public Map<String, Integer> mapToLocalProjects(@Nonnull String upstreamId, @Nonnull Iterable<String> upstreamProjectIds) {
        requireNonNull(upstreamId, "upstreamId");
        requireNonNull(upstreamProjectIds, "upstreamProjectIds");

        ImmutableMap.Builder<String, Integer> builder = ImmutableMap.builder();
        Iterable<AoProjectMapping> mappings = projectMappingDao.getByUpstreamId(upstreamId, upstreamProjectIds);
        for (AoProjectMapping mapping : mappings) {
            builder.put(mapping.getExternalId(), mapping.getLocalId());
        }
        return builder.build();
    }

    @Nonnull
    @Override
    public Map<String, Integer> mapToLocalRepositories(@Nonnull String upstreamId, @Nonnull Iterable<String> upstreamRepositoryIds) {
        requireNonNull(upstreamId, "upstreamId");
        requireNonNull(upstreamRepositoryIds, "upstreamRepositoryIds");

        ImmutableMap.Builder<String, Integer> builder = ImmutableMap.builder();
        Iterable<AoRepositoryMapping> mappings = repositoryMappingDao.getByUpstreamId(upstreamId, upstreamRepositoryIds);
        for (AoRepositoryMapping mapping : mappings) {
            builder.put(mapping.getExternalId(), mapping.getLocalId());
        }
        return builder.build();
    }

    @Nullable
    @VisibleForTesting
    String getUpstreamNamespaceFromProperties() {
        Optional<String> upstreamNamespace = config.getUpstreamNamespace();
        if (!upstreamNamespace.isPresent()) {
            return null;
        }

        String ns = upstreamNamespace.get();
        if (ns.length() > MAX_NAMESPACE_LENGTH || !ns.matches(Project.NAMESPACE_REGEXP)) {
            KeyedMessage message = i18nService.createKeyedMessage("bitbucket.mirroring.upstream.namespace.invalid", ns);
            addJohnsonEvent(message.getLocalisedMessage());
            throw new InvalidUpstreamServerNamespaceException(message);
        }

        return ns;
    }

    private void maybeClearProgress(String upstreamId) {
        inTransactionVoid(() -> {
            int syncedRepos = repositoryMappingDao.countSyncedInProgress(upstreamId);
            int totalRepos = repositoryMappingDao.countAllInProgress(upstreamId);
            if (syncedRepos >= totalRepos) {
                projectMappingDao.clearAllProgress(upstreamId);
            }
        });
    }

    private boolean isValidUrl(String baseUrl) {
        try {
            new URL(baseUrl);
            return true;
        } catch (MalformedURLException e) {
            return false;
        }
    }

    private InternalUpstreamServer updateUpstreamServerDetails(@Nonnull String upstreamId,
                                                               @Nonnull IntegrationState state,
                                                               @Nullable UpstreamAccount account) {

        requireNonNull(upstreamId, "upstreamId");
        requireNonNull(state, "state");

        return inTransaction(() -> {
            AoUpstreamServer upstream = setUpstreamServerIntegrationState(upstreamId, state);

            if (account != null) {
                // the upstream is cloud and the fetch of the account details succeeded
                upstream = setUpstreamServerAccountName(upstream, account.getName());
            }

            return upstream;
        });
    }

    private AoUpstreamServer setUpstreamServerIntegrationState(@Nonnull String upstreamId, @Nonnull IntegrationState state) {
        requireNonNull(upstreamId, "upstreamId");
        requireNonNull(state, "state");

        return inTransaction(() -> {
            AoUpstreamServer upstream = upstreamServerDao.getById(upstreamId);
            if (upstream == null) {
                throw throwNoSuchUpstream(upstreamId);
            }
            return setUpstreamServerIntegrationState(upstream, state);
        });
    }

    private AoUpstreamServer setUpstreamServerIntegrationState(@Nonnull AoUpstreamServer upstream,
                                                               @Nonnull IntegrationState state) {
        requireNonNull(upstream, "upstream");
        requireNonNull(state, "state");

        String upstreamId = upstream.getId();
        IntegrationState currentState = upstream.getState();
        if (currentState == state) {
            // nothing to do
            return upstream;
        }

        if (currentState == REMOVED && state != initialStateFor(upstream)) {
            // From REMOVED, you can only go back to INITIALIZING (server) or PENDING (cloud) when the command
            // is given to re-integrate the mirror. Any other state is bound to be an out-of-order DISABLED
            // event (DISABLED and REMOVED notifications are sent at nearly the same time)
            log.trace("Upstream state is REMOVED. Ignoring request to set the state to {}", state);
            return upstream;
        }

        if (state == UNKNOWN && (currentState == INITIALIZING || currentState == PENDING)) {
            // if we cannot determine the upstream state, we'll remain in the pending state / initializing state
            // since mirroring isn't active anyway
            log.trace("Upstream state is {}. Ignoring request to set the state to UNKNOWN", currentState);
            return upstream;
        }

        log.info("Setting upstream state for {} to {}", upstreamId, state);
        upstream = upstreamServerDao.transitionState(upstream, state);

        switch (state) {
            case INSTALLED:
                securityHelper.createNewOrActivateExistingUserForUpstream(upstreamId);
                eventPublisher.publish(new MirrorInstalledUpstreamEvent(DefaultUpstreamService.this, upstream));
                break;
            case REMOVED:
                securityHelper.deactivateUserForUpstream(upstreamId);
                eventPublisher.publish(new MirrorRemovedUpstreamEvent(DefaultUpstreamService.this, upstream));
                break;
        }

        return upstream;
    }

    private AoUpstreamServer setUpstreamServerAccountName(@Nonnull AoUpstreamServer upstream,
                                                          @Nonnull String accountName) {
        requireNonNull(upstream, "upstream");
        requireNonNull(accountName, "accountName");

        String existingAccountName = upstream.getAccount().map(UpstreamAccount::getName).orElse(null);
        if (!Objects.equals(accountName, existingAccountName)) {
            if (existingAccountName == null) {
                log.info("{}: setting upstream account name to {}", describe(upstream), accountName);
            } else {
                log.info("{}: upstream account name has changed from {} to {}", describe(upstream), existingAccountName, accountName);
            }
            upstreamServerDao.setAccountName(upstream.getId(), accountName);

            // if the account name was previously unset then this is the first time we know what it is
            // if there's no configured namespace then we should set the namespace to the account name
            // note: the configured namespace can be removed from bitbucket.properties but it's an insignificant edge-case
            if (existingAccountName == null && !config.getUpstreamNamespace().isPresent()) {
                String namespace = findFreeNamespaceFor(upstream, normalize(accountName));
                log.debug("{}: namespace is changing from {} to {} because the upstream account name is now known to be {}",
                        describe(upstream), upstream.getNamespace(), namespace, accountName);
                upstreamServerDao.setNamespace(upstream.getId(), namespace);
            }
        }
        return upstreamServerDao.getById(upstream.getId());
    }

    private void addJohnsonEvent(String message) {
        if (johnsoned) {
            return;
        }

        synchronized (this) {
            if (johnsoned) {
                return;
            }
            johnsoned = true;
            johnsonHelper.addJohnsonEvent("plugin-failed", message, EventLevel.ERROR);
        }
    }

    private void checkIsMirror() {
        if (!isMirror()) {
            throw notAMirrorException();
        }
    }

    /**
     * Creates an upstream using the passed upstream URL and type.
     *
     * @param upstreamUrl the URL for the upstream.
     * @param type the type of the upstream.  @return the upstream.
     * @throws IllegalStateException if an upstream already exists.
     */
    @Nonnull
    private InternalUpstreamServer createSingleUpstream(@Nonnull String upstreamUrl,
                                                        @Nonnull UpstreamServerType type) {
        return inTransaction(() -> {
            AoUpstreamServer upstream = upstreamServerDao.getUpstream();
            if (upstream != null) {
                throw new IllegalStateException(String.format("Upstream already registered as '%s'.", upstream.getBaseUrl()));
            }

            return createUpstreamInternal(upstreamUrl,
                    config.getUpstreamNamespace().orElseGet(() -> generateNamespaceForUrl(upstreamUrl)), type);
        });
    }

    /**
     * Creates a {@link InternalUpstreamServer} even if one currently exists. This method must be called from inside
     * a transaction.
     *
     * @param baseUrl the URL of the upstream server.
     * @param namespace the namespace to use in clone URLs for repositories mirrored from the upstream server.
     * @return the new {@code InternalUpstreamServer}.
     */
    private InternalUpstreamServer createUpstreamInternal(@Nonnull String baseUrl, @Nonnull String namespace,
                                                          @Nonnull UpstreamServerType type) {
        String id = UUID.nameUUIDFromBytes(baseUrl.getBytes(StandardCharsets.UTF_8)).toString().toLowerCase(Locale.ROOT);

        log.debug("Creating new upstream server in database with ID {}, baseUrl {}, namespace {}, type {}",
                id, baseUrl, namespace, type);

        CreateUpstreamServerRequest.Builder builder = new CreateUpstreamServerRequest.Builder()
                .id(id)
                .baseUrl(baseUrl)
                .namespace(namespace)
                .upstreamType(type)
                .state(initialStateFor(type));

        return upstreamServerDao.create(builder.build());
    }

    @Nonnull
    private String getUpstreamUrlFromPropertiesOrJohnson() {
        Optional<String> upstreamUrl = config.getServerUpstreamUrl();
        if (!upstreamUrl.isPresent()) {
            KeyedMessage message = i18nService.createKeyedMessage("bitbucket.mirroring.upstream.url.not.configured");
            addJohnsonEvent(message.getLocalisedMessage());
            throw new InvalidUpstreamServerUrlException(message);
        }

        String url = upstreamUrl.get();
        if (!isValidUrl(url)) {
            KeyedMessage message = i18nService.createKeyedMessage("bitbucket.mirroring.upstream.url.invalid", url);
            addJohnsonEvent(message.getLocalisedMessage());
            throw new InvalidUpstreamServerUrlException(message);
        }

        return upstreamUrl.get();
    }

    @Nonnull
    private UpstreamServerType getUpstreamTypePropertyOrJohnson() throws InvalidUpstreamTypeException {
        Optional<UpstreamServerType> serverType = config.getUpstreamType();
        if (!serverType.isPresent()) {
            KeyedMessage message = i18nService.createKeyedMessage("bitbucket.mirroring.upstream.type.not.configured");
            addJohnsonEvent(message.getLocalisedMessage());
            throw new InvalidUpstreamTypeException(message);
        }

        UpstreamServerType type = serverType.get();
        // reject CLOUD if CLOUD_SUPPORT is disabled
        if (type == UpstreamServerType.BITBUCKET_CLOUD && !featureManager.isEnabled(CloudMirroringFeature.CLOUD_SUPPORT)) {
            KeyedMessage message = i18nService.createKeyedMessage("bitbucket.mirroring.upstream.type.invalid", String.valueOf(serverType));
            addJohnsonEvent(message.getLocalisedMessage());
            throw new InvalidUpstreamTypeException(message);
        }

        return type;
    }

    private InternalUpstreamServer getOrCreateUpstreamServer() {
        return transactionTemplate.execute(() -> {
            InternalUpstreamServer upstreamServer = upstreamServerDao.getUpstream();
            if (upstreamServer == null) {
                String upstreamUrl;
                UpstreamServerType type = getUpstreamTypePropertyOrJohnson();
                if (type == BITBUCKET_CLOUD) {
                    upstreamUrl = getCloudUpstreamUrlOrJohnson();
                } else if (type == BITBUCKET_SERVER) {
                    upstreamUrl = getUpstreamUrlFromPropertiesOrJohnson();
                } else {
                    throw new IllegalStateException(String.format("Unexpected type '%s'.", type));
                }
                String upstreamNamespace = getUpstreamNamespaceFromProperties();
                if (upstreamNamespace == null) {
                    upstreamNamespace = generateNamespaceForUrl(upstreamUrl);
                }
                return createUpstreamInternal(upstreamUrl, upstreamNamespace, type);
            } else {
                return upstreamServer;
            }
        });
    }

    private AoUpstreamServer getUpstreamOrFail(String upstreamId) {
        if (!isMirror()) {
            throw notAMirrorException();
        }

        AoUpstreamServer upstream = inTransaction(() -> upstreamServerDao.getById(upstreamId));
        if (upstream == null) {
            throw throwNoSuchUpstream(upstreamId);
        }
        return upstream;
    }

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

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

    private boolean isAuthenticatedAsUpstream(String upstreamId) {
        return MirroringUserUtils.isUpstreamUserFor(authenticationContext.getCurrentUser(), upstreamId);
    }

    private void onStateUpdated(@Nonnull String upstreamId, IntegrationState expectedState) {
        requireNonNull(upstreamId, "upstreamId");

        if (isAuthenticatedAsUpstream(upstreamId)) {
            setUpstreamServerIntegrationState(upstreamId, expectedState);
        } else {
            refreshStateFromUpstream(getUpstreamOrFail(upstreamId), REFRESH_DELAY_MS);
        }
    }

    private void refreshStateFromUpstream(@Nonnull UpstreamServer upstream, long delayMs) {
        if (delayMs <= 0L) {
            refreshFromUpstreamNow(upstream).claim();
        } else {
            executorService.schedule(new DelayedRefreshStateCommand(upstream), delayMs, TimeUnit.MILLISECONDS);
        }
    }

    private Promise<UpstreamServer> refreshFromUpstreamNow(@Nonnull UpstreamServer upstream) {
        log.debug("Refreshing mirroring state with upstream {}", describe(upstream));

        // neither of these promises will fail due to recoveries on the promises that transform failures into valid
        // return values
        Promise<IntegrationState> upstreamState = refreshUpstreamState(upstream);
        // a no-op for server
        Promise<UpstreamAccount> upstreamAccount = refreshUpstreamAccount(upstream);

        return Promises.when(upstreamState, upstreamAccount)
                .map(accountAndState -> {
                    IntegrationState state = (IntegrationState) accountAndState.get(0);
                    UpstreamAccount account = (UpstreamAccount) accountAndState.get(1);

                    return updateUpstreamServerDetails(upstream.getId(), state, account);
                });
    }

    private NoSuchUpstreamException noSuchUpstreamException(String upstreamId) {
        return new NoSuchUpstreamException(i18nService.createKeyedMessage("bitbucket.mirroring.no.such.upstream.server", upstreamId));
    }

    private IncompatibleApplicationModeException notAMirrorException() {
        return new IncompatibleApplicationModeException(i18nService.createKeyedMessage("bitbucket.mirroring.not.a.mirror"));
    }

    private NoSuchUpstreamException notYetRegisteredException() {
        return new NoSuchUpstreamException(i18nService.createKeyedMessage("bitbucket.mirroring.upstream.not.yet.registered"));
    }

    private Promise<UpstreamAccount> refreshUpstreamAccount(@Nonnull UpstreamServer upstream) {
        Promise<UpstreamAccount> upstreamAccount;
        // only a cloud mirror that has completed the installation process (UNKNOWN implies it was last known to be
        // INSTALLED but we're currently failing to refresh that) will have the team ID available necessary to GET the
        // team account from the upstream
        if (upstream.getType() == BITBUCKET_CLOUD && (upstream.getState() == INSTALLED || upstream.getState() == UNKNOWN)) {
            upstreamAccount = upstreamClientFactory.createForCloud(upstream)
                    .getAccount()
                    .recover(throwable -> {
                        log.warn("Could not retrieve account details at upstream server {}", describe(upstream));
                        log.debug("Exception details:", throwable);
                        return null;
                    });
        } else {
            // only cloud upstreams have accounts
            upstreamAccount = Promises.promise(null);
        }
        return upstreamAccount;
    }

    private Promise<IntegrationState> refreshUpstreamState(@Nonnull UpstreamServer upstream) {
        return upstreamClientFactory.create(upstream)
                .isMirrorInstalled()
                .map(installed -> {
                    if (installed) {
                        return INSTALLED;
                    }
                    IntegrationState curState = upstream.getState();
                    return (curState == INITIALIZING || curState == PENDING || curState == REMOVED)
                            ? curState // not installed is expected in while in INITIALIZING, PENDING or REMOVED state
                            : REMOVED; // no installed while in INSTALLED and UNKNOWN state means removed
                })
                .recover(throwable -> {
                    log.warn("Could not retrieve mirror state at upstream server {}. Reason:", describe(upstream),
                            throwable);
                    IntegrationState curState = upstream.getState();
                    return (curState == INITIALIZING || curState == PENDING || curState == REMOVED)
                            // the request failed but we're in a state that we can never return to by querying the
                            // upstream. to avoid losing state information, we won't move to the UNKNOWN state.
                            ? curState
                            : UNKNOWN; // no installed while in INSTALLED and UNKNOWN state means removed
                });
    }

    private Promise<Void> registerUpstream(UpstreamServer upstream) {
        log.info("Registering with '{}' as a mirror server", describe(upstream));
        return upstreamClientFactory.createForServer(upstream)
                .registerAsMirror(new SimpleMirroringRequest.Builder()
                    .product(
                            abbreviate(getProductName(), MAX_PRODUCT_TYPE_LENGTH),
                            abbreviate(propertiesService.getBuildVersion(), MAX_PRODUCT_VERSION_LENGTH))
                    .mirrorServer(
                            requireNonNull(propertiesService.getServerId()),
                            abbreviate(config.getDisplayName(), MAX_MIRROR_NAME_LENGTH))
                    .mirrorBaseUrl(requireNonNull(propertiesService.getBaseUrl()).toASCIIString())
                    .addonDescriptor(URI.create(mirroringNavBuilder.rest().upstreamServer(upstream.getId()).addon().descriptor().getRelative()))
                    .build())
                .done(nothing -> inTransactionVoid(() -> upstreamServerDao.transitionState(upstream.getId(), PENDING)));
    }

    @VisibleForTesting
    String getProductName() {
        return Product.NAME;
    }

    private boolean settingsChangeNeedsFullSync(UpstreamSettingsChangedEvent event) {
        // there should always be settings but we handle this case anyway
        if (event.getOldSettings() == null) {
            return true;
        }

        // we're going from SELECTED_PROJECTS to ALL_PROJECTS
        if (event.getNewSettings().getMode() == MirroringMode.ALL_PROJECTS
                && event.getOldSettings().getMode() != MirroringMode.ALL_PROJECTS) {
            return true;
        }

        // mirrored project IDs have changed for SELECTED_PROJECTS
        return event.getNewSettings().getMode() == MirroringMode.SELECTED_PROJECTS
                && event.getOldSettings().getMode() == MirroringMode.SELECTED_PROJECTS
                && !event.getOldSettings().getMirroredProjectIds().equals(event.getNewSettings().getMirroredProjectIds());

    }

    private void startFullSynchronizationAsUpstreamUser(UpstreamServer upstream, SyncLevel syncLevel) {
        EscalatedSecurityContext securityContext = securityHelper.asUserForUpstream(upstream.getId());
        if (securityContext != null) {
            securityContext.call(() -> {
                mirrorSynchronizationHelper.startFullSynchronization(upstream, syncLevel);
                return null;
            });
        } else {
            log.warn("No user is defined for upstream '{}'. Skipping synchronization", describe(upstream));
        }
    }

    private ArgumentValidationException throwBadRequest(String key) {
        throw new ArgumentValidationException(i18nService.createKeyedMessage(key));
    }

    private AuthorisationException throwNotPermitted() {
        throw new AuthorisationException(i18nService.createKeyedMessage("bitbucket.mirroring.operation.not.allowed"));
    }

    private NoSuchUpstreamException throwNoSuchUpstream(String upstreamId) {
        throw noSuchUpstreamException(upstreamId);
    }

    private void validateAuthenticatedAsUpstream(String upstreamId) {
        if (!isAuthenticatedAsUpstream(upstreamId)) {
            throw throwNotPermitted();
        }
    }

    private static IntegrationState initialStateFor(@Nonnull UpstreamServerType type) {
        switch (type) {
            case BITBUCKET_CLOUD:
                return PENDING;
            case BITBUCKET_SERVER:
                return INITIALIZING;
            default:
                throw new IllegalArgumentException("Unexpected upstream server type: " + type);
        }
    }

    private String getCloudUpstreamUrlOrJohnson() {
        return getCloudUpstreamUrl().orElseThrow(() -> {
            KeyedMessage message = i18nService.createKeyedMessage("bitbucket.mirroring.upstream.cloud.url.invalid");
            addJohnsonEvent(message.getLocalisedMessage());
            return new InvalidUpstreamServerUrlException(message);
        });
    }

    private Optional<String> getCloudUpstreamUrl() {
        return config.getCloudUpstreamUrl()
                .map(StringUtils::stripToNull)
                .filter(this::isValidUrl);
    }

    private static IntegrationState initialStateFor(@Nonnull UpstreamServer upstream) {
        return initialStateFor(upstream.getType());
    }

    /**
     * Finds a free namespace for the supplied upstream server based on the supplied proposed namespace. If the proposed
     * namespace is already taken by another upstream an intensive search for a version of the proposal with "2", "3",
     * ... and so on appended is found. If the proposed or alternate namespace (from the search) is already taken by the
     * supplied upstream, it is returned (to allow this method to be used for both new upstreams and existing upstreams
     * which have had their account name changed).
     */
    @Nonnull
    private String findFreeNamespaceFor(@Nullable UpstreamServer upstream, @Nonnull String proposedNamespace) {
        AoUpstreamServer clash = upstreamServerDao.getByNamespace(proposedNamespace);
        String nonClashNamespace = proposedNamespace;
        int attempt = 0;
        while (clash != null && !clash.equals(upstream)) {
            String attemptStr = Integer.toString(attempt + 2); // "namespace2" if "namespace" is taken
            nonClashNamespace = proposedNamespace.substring(
                    0, Math.min(proposedNamespace.length(), MAX_NAMESPACE_LENGTH - attemptStr.length())) + attemptStr;
            clash = upstreamServerDao.getByNamespace(nonClashNamespace);
            attempt++;
        }
        return nonClashNamespace;
    }

    @Nonnull
    private String generateNamespaceForUrl(@Nonnull String upstreamUrl) {
        // Compute a namespace for a new upstream server based on the host name in its base URL.
        String namespace = URI.create(upstreamUrl).normalize().getHost().replaceFirst("\\..*", "");
        namespace = normalize(namespace);
        return findFreeNamespaceFor(null, namespace);
    }

    @Nonnull
    private String normalize(@Nonnull String namespace) {
        namespace = Normalizer.normalize(namespace, Normalizer.Form.NFKD).replaceAll("[^a-zA-Z0-9_\\-]", "_");
        if (namespace.length() > MAX_NAMESPACE_LENGTH) {
            namespace = namespace.substring(0, MAX_NAMESPACE_LENGTH);
        }
        return namespace.toLowerCase(Locale.ROOT);
    }

    @VisibleForTesting
    class DelayedRefreshStateCommand implements Callable<Void> {
        private final UpstreamServer upstream;

        public DelayedRefreshStateCommand(UpstreamServer upstream) {
            this.upstream = upstream;
        }

        @Override
        public Void call() throws Exception {
            refreshFromUpstreamNow(upstream).claim();
            return null;
        }
    }
}
