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

import com.atlassian.bitbucket.AuthorisationException;
import com.atlassian.bitbucket.auth.AuthenticationContext;
import com.atlassian.bitbucket.concurrent.RepositoryLock;
import com.atlassian.bitbucket.event.repository.RepositoryDeletedEvent;
import com.atlassian.bitbucket.event.user.UserCleanupEvent;
import com.atlassian.bitbucket.i18n.I18nService;
import com.atlassian.bitbucket.i18n.KeyedMessage;
import com.atlassian.bitbucket.permission.Permission;
import com.atlassian.bitbucket.permission.PermissionService;
import com.atlassian.bitbucket.project.*;
import com.atlassian.bitbucket.repository.*;
import com.atlassian.bitbucket.repository.ref.restriction.RefAccessRequest;
import com.atlassian.bitbucket.repository.ref.restriction.RefAccessType;
import com.atlassian.bitbucket.repository.ref.restriction.RefRestrictionService;
import com.atlassian.bitbucket.repository.sync.*;
import com.atlassian.bitbucket.scm.Command;
import com.atlassian.bitbucket.scm.git.GitScm;
import com.atlassian.bitbucket.user.ApplicationUser;
import com.atlassian.bitbucket.user.SecurityService;
import com.atlassian.bitbucket.util.*;
import com.atlassian.cache.Cache;
import com.atlassian.cache.CacheFactory;
import com.atlassian.cache.CacheSettingsBuilder;
import com.atlassian.event.api.EventListener;
import com.atlassian.event.api.EventPublisher;
import com.atlassian.sal.api.transaction.TransactionCallback;
import com.atlassian.sal.api.transaction.TransactionTemplate;
import com.atlassian.stash.internal.repository.sync.command.*;
import com.atlassian.stash.internal.repository.sync.dao.RefSyncDao;
import com.atlassian.stash.internal.repository.sync.dao.RefSyncNotConfiguredException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.util.Map;

import static com.atlassian.stash.internal.repository.sync.SimpleRejectedRef.isTag;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;

/**
 * @since 2.6
 */
@SuppressWarnings("ConstantConditions")
public class DefaultRefSyncService implements InternalRefSyncService {

    public static final ProjectVisitor<ApplicationUser> TO_OWNER = new AbstractProjectVisitor<ApplicationUser>() {

        @Override
        public ApplicationUser visit(@Nonnull PersonalProject project) {
            return project.getOwner();
        }
    };

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

    private final AuthenticationContext authenticationContext;
    private final RefSyncConfig config;
    private final EventPublisher eventPublisher;
    private final I18nService i18nService;
    private final PermissionService permissionService;
    private final RefSyncDao refSyncDao;
    private final RefSyncCommandFactory refSyncCommandFactory;
    private final RefRestrictionService refRestrictionService;
    private final RefService refService;
    private final RepositoryService repositoryService;
    private final SecurityService securityService;

    private final Cache<Integer, Boolean> settingsCache;
    private final TransactionTemplate transactionTemplate;

    public DefaultRefSyncService(AuthenticationContext authenticationContext, RefSyncConfig config,
                                 EventPublisher eventPublisher, I18nService i18nService,
                                 PermissionService permissionService, RefSyncCommandFactory refSyncCommandFactory,
                                 RefSyncDao refSyncDao, RefRestrictionService refRestrictionService,
                                 RefService refService, RepositoryService repositoryService,
                                 SecurityService securityService, TransactionTemplate transactionTemplate,
                                 CacheFactory cacheFactory) {
        this.authenticationContext = authenticationContext;
        this.config = config;
        this.eventPublisher = eventPublisher;
        this.i18nService = i18nService;
        this.permissionService = permissionService;
        this.refSyncDao = refSyncDao;
        this.refSyncCommandFactory = refSyncCommandFactory;
        this.refRestrictionService = refRestrictionService;
        this.refService = refService;
        this.repositoryService = repositoryService;
        this.securityService = securityService;
        this.transactionTemplate = transactionTemplate;

        settingsCache = cacheFactory.getCache("com.atlassian.bitbucket.repository.sync.RefSyncService.settingsCache",
                this::isEnabled,
                new CacheSettingsBuilder()
                        .remote()
                        .replicateAsynchronously()
                        .replicateViaInvalidation()
                        .build()
        );
    }

    @Override
    public RefSyncStatus getStatus(@Nonnull Repository repository) {
        checkNotNull(repository, "repository");

        RefSyncStatus status = transactionTemplate.execute(() -> refSyncDao.getStatus(repository.getId()));

        if (status == null) {
            return null;
        }
        //The database doesn't know whether syncing is available or not, so that must be added after the fact
        return new SimpleRefSyncStatus.Builder(status)
                .available(isAvailable(repository))
                .build();
    }

    @Override
    public RejectedRef getStatus(@Nonnull Repository repository, @Nonnull String refId) {
        checkNotNull(repository, "repository");
        checkArgument(!checkNotNull(refId, "refId").trim().isEmpty(), "A non-blank refId is required");

        return transactionTemplate.execute(() -> refSyncDao.getRef(repository.getId(), refId));
    }

    @Override
    public boolean isAvailable(@Nonnull Repository repository) {
        checkNotNull(repository, "repository");
        if (!config.isPossible()) {
            //The system configuration does not support ref synchronization.
            return false;
        }
        if (!GitScm.ID.equals(repository.getScmId())) {
            //Synchronization is only supported for git repositories
            return false;
        }

        Repository origin = repository.getOrigin();
        if (origin == null) {
            //Only forks can have synchronization enabled. Top-level repositories have nothing to synchronise with
            return false;
        }

        Project project = repository.getProject();
        if (project.getType() == ProjectType.PERSONAL) {
            //For personal forks, synchronization is only available so long as the owning user can still see the fork's
            //origin. Note that it is the _owner's_ permission being checked here, _not_ the current user's
            return permissionService.hasRepositoryPermission(project.accept(TO_OWNER), origin, Permission.REPO_READ);
        }

        //Otherwise, if the repository is not a personal fork, it's always synchronizable
        return true;
    }

    @Override
    public boolean isEnabled(@Nonnull Repository repository) {
        checkNotNull(repository, "repository");
        return settingsCache.get(repository.getId());
    }

    @EventListener
    public void onRepositoryDeleted(@Nonnull RepositoryDeletedEvent event) {
        Integer repositoryId = event.getRepository().getId();

        transactionTemplate.execute(() -> {
            refSyncDao.cleanup(repositoryId);

            return null;
        });
    }

    @EventListener
    public void onUserCleanup(@Nonnull UserCleanupEvent event) {
        transactionTemplate.execute(new BulkDisableOperation(event.getDeletedUser()));
    }

    @Override
    public void setEnabled(@Nonnull Repository repository, boolean enabled) {
        checkNotNull(repository, "repository");
        if (!permissionService.hasRepositoryPermission(repository, Permission.REPO_ADMIN)) {
            throw new AuthorisationException(i18nService.createKeyedMessage("bitbucket.service.repository.refsync.notanadmin"));
        }
        if (enabled && !isAvailable(repository)) {
            if (repository.isFork()) {
                //This will need to be kept in sync with the implementation of isAvailable. Currently, for forks, the
                //only way isAvailable will return false is in the case of a personal fork belonging to a user who has
                //lost access to its upstream. To be more clear for callers, the exception message here reflects that.
                ApplicationUser owner = repository.getProject().accept(TO_OWNER);
                Repository upstream = repository.getOrigin();

                throw new RefSyncNotAvailableException(i18nService.createKeyedMessage(
                        "bitbucket.service.repository.refsync.notavailable.detail",
                        repository.getProject().getKey(), repository.getSlug(),
                        owner.getDisplayName(), owner.getName(),
                        upstream.getProject().getKey(), upstream.getSlug()));
            }

            throw new RefSyncNotAvailableException(i18nService.createKeyedMessage("bitbucket.service.repository.refsync.notafork",
                    repository.getProject().getKey(), repository.getSlug()));
        }

        time(new SetEnabledAndSynchronizeOperation(repository, enabled));
    }

    @Override
    public void synchronize(@Nonnull RefSyncRequest request) {
        checkNotNull(request, "request");

        ApplicationUser user = authenticationContext.getCurrentUser();
        if (user == null) {
            throw new AuthorisationException(
                    i18nService.createKeyedMessage("bitbucket.service.repository.refsync.anonymous"));
        }

        Repository repository = request.getRepository();
        String refId = checkNotNull(request.getRefId(), "refId");
        if (!permissionService.hasRepositoryPermission(user, repository, Permission.REPO_WRITE)) {
            throw new AuthorisationException(i18nService.createKeyedMessage(
                    "bitbucket.service.repository.refsync.readonly",
                    repository.getProject().getKey(), repository.getSlug()));
        }

        Ref ref = refService.resolveRef(new ResolveRefRequest.Builder(repository)
                .refId(refId)
                .build());
        if (ref == null) {
            KeyedMessage message = i18nService.createKeyedMessage("bitbucket.service.repository.refsync.nosuchref",
                    repository.getProject().getKey(), repository.getSlug(), refId);
            if (isTag(refId)) {
                throw new NoSuchTagException(message, refId);
            }
            throw new NoSuchBranchException(message, refId);
        }
        RefAccessRequest accessRequest = new RefAccessRequest.Builder(request.getRepository(), RefAccessType.UPDATE)
                .ref(ref)
                .build();
        if (!refRestrictionService.hasPermission(accessRequest)) {
            throw new AuthorisationException(i18nService.createKeyedMessage(
                    "bitbucket.service.repository.refsync.unauthorizedref",
                    repository.getProject().getKey(), repository.getSlug(), refId));
        }
        if (!isEnabled(repository)) {
            throw new RefSyncNotEnabledException(i18nService.createKeyedMessage(
                    "bitbucket.service.repository.refsync.notenabled",
                    repository.getProject().getKey(), repository.getSlug()));
        }
        if (!isAvailable(repository)) {
            throw new RefSyncNotAvailableException(i18nService.createKeyedMessage(
                    "bitbucket.service.repository.refsync.notavailable",
                    repository.getProject().getKey(), repository.getSlug()));
        }

        RejectedRef refStatus = getStatus(repository, refId);
        if (refStatus == null) {
            throw new RefAlreadySynchronizedException(i18nService.createKeyedMessage(
                    "bitbucket.service.repository.refsync.alreadyinsync",
                    repository.getProject().getKey(), repository.getSlug(), refId));
        }

        RefSyncAction action = request.getAction();
        AbstractOperation<?> operation;
        switch (action) {
            case MERGE:
                if (refStatus.getState() != RejectedRefState.DIVERGED) {
                    throw new RefAlreadySynchronizedException(i18nService.createKeyedMessage(
                            "bitbucket.service.repository.refsync.nothingtomerge",
                            repository.getProject().getKey(), repository.getSlug(), refId));
                }
                operation = new MergeOperation(repository, refId, user, request.getContext());

                break;
            case DISCARD:
                operation = new DiscardOperation(repository, refId, user);

                break;
            default:
                throw new UnsupportedOperationException(
                        i18nService.getMessage("bitbucket.service.repository.refsync.actionnotimplemented", action));
        }

        time(operation);
    }

    @Override
    public void synchronize(@Nonnull BulkRefSyncRequest request) {
        checkNotNull(request, "request");

        time(new SynchronizeOperation(request));
    }

    @Override
    public void updateStatus(@Nonnull BulkRefSyncRequest request) {
        checkNotNull(request, "request");

        time(new ReviewOperation(request));
    }

    private static void time(UncheckedOperation<?> operation) {
        TimerUtils.time(operation.toString(), operation);
    }

    private boolean isEnabled(int repositoryId) {
        return transactionTemplate.execute(() -> refSyncDao.isEnabled(repositoryId));
    }

    private void updateStatus(Repository repository, RefSyncResult result) {
        transactionTemplate.execute(() -> {
            try {
                refSyncDao.setStatus(repository.getId(), result);
            } catch (RefSyncNotConfiguredException e) {
                throw new IllegalStateException("Ref synchronization is not configured for " +
                        repository.getProject().getKey() + "/" + repository.getSlug() +
                        "; status cannot be updated");
            }

            return null;
        });
    }

    /**
     * A base for building {@link com.atlassian.stash.internal.repository.sync.command.AbstractCommandParameters parameter}-based operations which automatically profiles
     * how long the operation takes.
     *
     * @param <T> The type of {@link com.atlassian.stash.internal.repository.sync.command.AbstractCommandParameters parameters} used by the operation.
     */
    private abstract class AbstractOperation<T extends AbstractCommandParameters>
            implements UncheckedOperation<Void> {

        protected final T parameters;

        protected AbstractOperation(T parameters) {
            this.parameters = parameters;
        }

        @Override
        public Void perform() {
            doPerform();

            return null;
        }

        protected abstract void doPerform();

        protected void maybeRaiseFor(@Nullable RefChange change) {
            if (change != null) {
                raise(new RefsSynchronizedEvent(DefaultRefSyncService.this,
                        parameters.getRepository(), parameters.getUpstream(), change));
            }
        }

        protected void raise(@Nonnull final RefsSynchronizedEvent event) {
            //Preconditions:
            //- StashAuthenticationContext.getCurrentUser() contains the user that should be on the event
            //- The current user has REPO_WRITE access to the repository
            //
            //Synchronization in response to RepositoryRefsChangedEvents promotes to REPO_WRITE, ensuring these
            //preconditions are met.
            //Manual synchronization explicitly verifies REPO_WRITE access for the current user (since they will
            //be updating refs), ensuring the preconditions are met.
            eventPublisher.publish(event);
        }
    }

    private class BulkDisableOperation
            implements PageProvider<Repository>, TransactionCallback<Void>, UncheckedOperation<Void> {

        private final ApplicationUser user;

        private BulkDisableOperation(ApplicationUser user) {
            this.user = user;
        }

        @Override
        public Void doInTransaction() {
            //This ensures, in perform(), that we can see all of the user's personal projects
            return securityService.withPermission(Permission.REPO_READ, "Bulk disabling ref synchronization").call(this);
        }

        @Nonnull
        @Override
        public Page<Repository> get(@Nonnull PageRequest request) {
            return repositoryService.findByOwner(user, request);
        }

        @Override
        public Void perform() {
            //Spin over all the user's repositories and, for any that are forks, ensure synchronization is disabled.
            //This does _not_ check if synchronization is enabled first; the extra database query to determine that
            //would just make this slower
            for (Repository repository : new PagedIterable<>(this, 25)) {
                if (repository.isFork()) {
                    refSyncDao.setEnabled(repository.getId(), false); //Will not create rows unnecessarily
                }
            }

            return null;
        }
    }

    /**
     * Discards local changes to a given ref in favour of its upstream changes. When this operation completes the
     * ref will <i>always</i> be in sync (where it may be "in sync" because it no longer exists).
     */
    private class DiscardOperation extends AbstractOperation<DiscardRefCommandParameters> {

        private DiscardOperation(Repository repository, String refId, ApplicationUser user) {
            super(new DiscardRefCommandParameters(repository, repository.getOrigin(), user, refId));
        }

        @Override
        public String toString() {
            return parameters.getRepository().getId() + ": Discard " + parameters.getRefId() +
                    " for " + parameters.getUpstream().getId();
        }

        @Override
        protected void doPerform() {
            RefChange change = refSyncCommandFactory.discardRef(parameters).call();
            maybeRaiseFor(change);

            transactionTemplate.execute(() -> {
                refSyncDao.removeRef(parameters.getRepository().getId(), parameters.getRefId());

                return null;
            });
        }
    }

    /**
     * Merges in changes from upstream to update a given ref from {@link RejectedRefState#DIVERGED diverged} to
     * strictly {@link RejectedRefState#AHEAD ahead}. When this operation completes the ref will still be out of
     * sync, but there will be no changes available upstream which it does not include.
     */
    private class MergeOperation extends AbstractOperation<MergeRefCommandParameters> {

        private MergeOperation(Repository repository, String refId, ApplicationUser user, Map<String, Object> context) {
            super(new MergeRefCommandParameters(repository, repository.getOrigin(),
                    user, refId, (String) context.get("commitMessage")));
        }

        @Override
        public String toString() {
            return parameters.getRepository().getId() + ": Merge " + parameters.getRefId() +
                    " from " + parameters.getUpstream().getId();
        }

        @Override
        protected void doPerform() {
            Command<RefChange> command = refSyncCommandFactory.mergeRef(parameters);
            command.setExecutionTimeout(config.getMergeTimeout());
            command.setIdleTimeout(config.getMergeTimeout());

            RefChange change = command.call();
            maybeRaiseFor(change);

            transactionTemplate.execute(() -> {
                refSyncDao.updateRef(parameters.getRepository().getId(), parameters.getRefId(), RejectedRefState.AHEAD);

                return null;
            });
        }
    }

    /**
     * Reviews the updated refs defined in the {@link SynchronizeRefsCommandParameters parameters} to update their
     * synchronization status.
     * <p>
     * This operation ensures that {@link SynchronizeOperation synchronization} is not running concurrently with
     * review, for a given repository, which may result in inconsistent state in the database. This operation can
     * never result in inconsistent repository state, but if synchronization happened to be running concurrently
     * it might:
     * <ul>
     * <li>Calculate incorrect state, depending on which refs were in the process of being synchronized, or</li>
     * <li>Produce inconsistent database state due to a race condition in updating the database</li>
     * </ul>
     */
    private class ReviewOperation extends AbstractOperation<ReviewRefsCommandParameters> {

        private ReviewOperation(BulkRefSyncRequest request) {
            super(new ReviewRefsCommandParameters(request));
        }

        @Override
        public String toString() {
            return parameters.getRepository().getId() + ": Synchronization review";
        }

        @Override
        protected void doPerform() {
            //No RefsSynchronizedEvent should be raised after a review; it is simply looking at the current state of
            //refs within the repository, not updating them
            RefSyncResult result = refSyncCommandFactory.reviewRefs(parameters).call();
            if (!(result == null || result.isEmpty())) {
                updateStatus(parameters.getRepository(), result);
            }
        }
    }

    /**
     * Uses the {@link RefSyncDao} to enable or disable synchronization and, if enabled, triggers a full
     * {@link SynchronizeOperation synchronization} for the repository.
     * <p>
     * This operation exists because, whether disabling or enabling synchronization, it must be done while
     * holding the {@link RepositoryLock lock}.
     * <ul>
     * <li>Disabling synchronization deletes all of the {@link RejectedRef rejected ref} rows, which may
     * result in inconsistent database state if incremental synchronization was running at the same time</li>
     * <li>Enabling synchronization immediately synchronizes all refs, potentially inserting some number
     * of {@link RejectedRef rejected refs}, which may result in inconsistent database state if a push
     * triggers incremental synchronization at the same time</li>
     * </ul>
     * <b>Note</b>: This operation is, itself, not profiled. If it triggers synchronization, that operation will
     * be profiled. Updating the database to enable or disable synchronization is not timed.
     */
    private class SetEnabledAndSynchronizeOperation implements UncheckedOperation<Void> {

        private final boolean enabled;
        private final Repository repository;

        private SetEnabledAndSynchronizeOperation(Repository repository, boolean enabled) {
            this.enabled = enabled;
            this.repository = repository;
        }

        @Override
        public Void perform() {
            //Update the repository's configuration to enable or disable synchronization. This must be done while
            //holding the lock because, if disabling, it may
            transactionTemplate.execute(() -> {
                refSyncDao.setEnabled(repository.getId(), enabled);

                return null;
            });

            //Using a replicateViaInvalidation cache means we can't put(...) the new value. Instead, we have to
            //remove the old value, to invalidate it on all nodes, and then let the next get load it again
            settingsCache.remove(repository.getId());

            if (enabled) {
                //If we've just enabled synchronization for a given repository, immediately synchronized _all_ refs
                //with its upstream one time. This simplifies the work required for incremental changes to keep the
                //two in sync over time by allowing us to make some logical assumptions about the state of refs
                time(new SynchronizeOperation(repository, authenticationContext.getCurrentUser()));
            }

            return null;
        }

        @Override
        public String toString() {
            return repository.getId() + ": Enabling synchronization";
        }
    }

    /**
     * Synchronizes refs in a given repository from its upstream. This operation is used for both full and incremental
     * synchronization operations, based on {@link SynchronizeRefsCommandParameters#isFull() the parameters}.
     * <p>
     * This operation ensures that synchronizing the refs in the repository on disk is paired with setting their status
     * in the database to guarantee consistent state across both.
     */
    private class SynchronizeOperation extends AbstractOperation<SynchronizeRefsCommandParameters> {

        private SynchronizeOperation(BulkRefSyncRequest request) {
            super(new SynchronizeRefsCommandParameters(request));
        }

        private SynchronizeOperation(Repository repository, ApplicationUser user) {
            super(new SynchronizeRefsCommandParameters(repository, repository.getOrigin(), user));
        }

        @Override
        public String toString() {
            StringBuilder builder = new StringBuilder(64).append(parameters.getRepository().getId())
                    .append(": ");
            if (parameters.isFull()) {
                builder.append("Full synchronization");
            } else {
                builder.append("Incremental synchronization for ")
                        .append(parameters.size())
                        .append(" refs");
            }
            return builder.toString();
        }

        @Override
        protected void doPerform() {
            RefSyncResult result = refSyncCommandFactory.synchronizeRefs(parameters).call();
            if (result == null) {
                return;
            }
            log.debug("{}: Synchronized {}/{} ref(s) from {} ({} rejected)",
                    parameters.getRepository(), result.getChangedRefs().size(), parameters.size(),
                    parameters.getUpstream(), result.getRejectedRefs().size());

            if (!result.isEmpty()) {
                //If at least one ref was changed or rejected, the RefsSynchronizedEvent is raised. Note that, even
                //if all refs were rejected, this event may be raised without any changed refs, as the event still
                //includes information about rejected refs that a consumer could, in theory, be interested in, and
                //consumers of RepositoryRefsChangedEvent must already handle the case where the event includes no
                //changed refs because remotely-merged pull requests trigger such events as well
                raise(new RefsSynchronizedEvent(DefaultRefSyncService.this, parameters.getRepository(),
                        result.getChangedRefs(), result.getRejectedRefs(), parameters.getUpstream(), parameters));
            }
            //After raising the event, update rejected refs for this repository. This uses the changed and rejected refs
            //from the current RefSyncResult to update _specific_ refs, allowing refs that were previously rejected to
            //be removed when they're successfully synchronized, while not blindly removing all rejected refs after each
            //synchronization (which is necessary since most synchronizations are incremental)
            updateStatus(parameters.getRepository(), result);
        }
    }
}
