package com.atlassian.stash.internal.pull.rescope;

import com.atlassian.bitbucket.pull.PullRequest;
import com.atlassian.bitbucket.pull.PullRequestRef;
import com.atlassian.bitbucket.pull.PullRequestSearchRequest;
import com.atlassian.bitbucket.pull.PullRequestState;
import com.atlassian.bitbucket.repository.*;
import com.atlassian.bitbucket.util.*;
import com.atlassian.bitbucket.util.Timer;
import com.atlassian.stash.internal.InternalConverter;
import com.atlassian.stash.internal.pull.InternalPullRequestService;
import com.atlassian.stash.internal.pull.InternalRescopeRequest;
import com.atlassian.stash.internal.pull.RescopeRequestDao;
import com.atlassian.stash.internal.repository.InternalRepository;
import com.atlassian.stash.internal.scm.InternalScmService;
import com.atlassian.bitbucket.scm.pull.MinimalPullRequest;
import com.atlassian.bitbucket.scm.pull.RepositoryRescopeCommandParameters;
import com.atlassian.bitbucket.scm.pull.RepositoryRescopeContext;
import com.atlassian.bitbucket.user.ApplicationUser;
import com.atlassian.bitbucket.auth.AuthenticationContext;
import com.atlassian.bitbucket.permission.Permission;
import com.atlassian.bitbucket.user.SecurityService;
import com.atlassian.stash.internal.user.InternalApplicationUser;
import com.google.common.base.Function;
import com.google.common.collect.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Nonnull;
import java.util.*;
import java.util.stream.Collectors;

import static com.atlassian.bitbucket.util.MoreCollectors.toImmutableList;
import static com.atlassian.stash.internal.InternalConverter.convertToInternalRepository;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;

/**
 * The process of calculating which pull requests to rescope, and how, is delegated to the SCM for the affected
 * repository. This is intended to allow the SCM to perform the necessary calculations in the most efficient manner
 * possible, potentially leveraging functionality which is not exposed to by normal APIs or SPIs. After the rescopes
 * have been calculated by the SCM, this component uses the
 * {@link com.atlassian.stash.internal.pull.InternalPullRequestService pull request service} to update the database and
 * raise related events.
 * <p>
 * This is implemented as a separate component to simplify transaction management, since {@code Transactional}
 * annotations are not applied for event listeners.
 *
 * @since 1.3
 */
@SuppressWarnings("ConstantConditions")
public class LegacyPullRequestRescopeService implements InternalPullRequestRescopeService {

    public static final int MAX_RESCOPE_ATTEMPTS = 3;

    private static final Function<PullRequest, MinimalPullRequest> MINIMIZER = new Function<PullRequest, MinimalPullRequest>() {

        @Override
        public SimpleMinimalPullRequest apply(PullRequest pullRequest) {
            return new SimpleMinimalPullRequest(pullRequest);
        }
    };

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

    //NOTE: None of these strategies should be run within the TransactionTemplate. They should not have any
    //      existing transaction around them.
    private final RescopeUpdateStrategy decline = new RescopeUpdateStrategy() {

        @Override
        public void apply(SimpleMinimalPullRequest pullRequest) {
            pullRequestService.decline(pullRequest.globalId, pullRequest.version);
        }

        @Override
        public String getErrorFormat() {
            return "Could not decline pull request %1$s";
        }
    };
    private final RescopeUpdateStrategy merge = new RescopeUpdateStrategy() {

        @Override
        public void apply(SimpleMinimalPullRequest pullRequest) {
            pullRequestService.closeMerged(pullRequest.globalId, pullRequest.version);
        }

        @Override
        public String getErrorFormat() {
            return "Could not mark pull request %1$s as merged";
        }
    };
    private final RescopeUpdateStrategy rescope = new RescopeUpdateStrategy() {

        @Override
        public void apply(SimpleMinimalPullRequest pullRequest) {
            pullRequestService.updateScope(pullRequest.globalId, pullRequest.version,
                    pullRequest.newFromHash, pullRequest.newToHash);
        }

        @Override
        public String getErrorFormat() {
            return "Could not rescope pull request %1$s";
        }
    };

    private final AuthenticationContext authenticationContext;
    private final RescopeRequestDao dao;
    private final InternalPullRequestService pullRequestService;
    private final InternalScmService scmService;
    private final SecurityService securityService;

    public LegacyPullRequestRescopeService(AuthenticationContext authenticationContext,
                                           RescopeRequestDao dao,
                                           InternalPullRequestService pullRequestService,
                                           InternalScmService scmService, SecurityService securityService) {
        this.authenticationContext = authenticationContext;
        this.dao = dao;
        this.pullRequestService = pullRequestService;
        this.scmService = scmService;
        this.securityService = securityService;
    }

    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void createRequest(@Nonnull InternalRepository repository, @Nonnull InternalApplicationUser user,
                              @Nonnull Collection<RefChange> changes, @Nonnull Date date) {
        dao.create(new InternalRescopeRequest.Builder(repository, user)
                .changes(changes)
                .createdDate(date)
                .build());
    }

    @Nonnull
    @Override
    @PreAuthorize("hasRepositoryPermission(#repository, 'REPO_ADMIN')")
    @Transactional
    public RepositoryRescopeResult rescope(@Nonnull Repository repository, @Nonnull List<String> branches) {
        InternalApplicationUser user = InternalConverter.convertToInternalUser(authenticationContext.getCurrentUser());
        InternalRepository repo = convertToInternalRepository(repository);
        List<RefChange> changes = branches.stream().map(MarkerRefChange::new).collect(Collectors.toList());

        dao.create(new InternalRescopeRequest.Builder(repo, user).changes(changes).build());
        return rescope(repository);
    }

    @Nonnull
    @Override
    @Transactional
    public RepositoryRescopeResult rescope(@Nonnull final Repository repository) {
        List<InternalRescopeRequest> requests = dao.findByRepository(repository.getId());
        if (requests.isEmpty()) {
            log.debug("{}: All rescope requests have already been processed", repository);
        }
        final Map<String, InternalApplicationUser> branchesToUsers = mapBranchesToUsers(requests);

        try (Timer ignored = TimerUtils.start("Rescoping pull requests for " + repository.toString())) {
            //Heighten privileges to read pull requests and later impersonate user that pushed
            RepositoryRescopeResult result = securityService.withPermission(Permission.REPO_READ, "Rescoping pull requests")
                    .call(new UncheckedOperation<RepositoryRescopeResult>() {
                        @Override
                        public RepositoryRescopeResult perform() {
                            int attempt = 1;
                            //Retry rescoping the repository if some pull requests failed to rescope due to errors.
                            //These are typically optimistic locking issues where the pull request was updated while the
                            //rescope was in progress. Locked pull requests are skipped by attemptUpdatePullRequestFor
                            //and should not trigger an immediate retry. The lock is expected to persist for a bit
                            //longer. The caller should retry after a delay.
                            RepositoryRescopeResult result;
                            do {
                                result = attemptUpdatePullRequestsFor(repository, branchesToUsers, attempt);
                            } while (result.hasErrors() && attempt++ < MAX_RESCOPE_ATTEMPTS);

                            if (log.isDebugEnabled()) {
                                log.debug("{}: Processed {} rescope requests: {}", repository, requests.size(),
                                        requests.stream().map(InternalRescopeRequest.TO_ID).collect(toImmutableList()));
                            }

                            return result;
                        }
                    });

            if (!result.isDone()) {
                branchesToUsers.entrySet().stream()
                        .map(entry -> {
                            InternalRescopeRequest.Builder builder =
                                    new InternalRescopeRequest.Builder(convertToInternalRepository(repository), entry.getValue());
                            if (entry.getKey() != null) {
                                builder.changes(ImmutableList.of(new MarkerRefChange(entry.getKey())));
                            }
                            return builder.build();
                        })
                        .forEach(dao::create);
            }

            dao.deleteByIds(requests.stream()
                    .map(InternalRescopeRequest::getId)
                    .collect(toImmutableList()));
            return result;
        }
    }

    /**
     * @param repository the repository
     * @param branchesToUsers map of branch ID (or {@code null} for all branches) to user that pushed to the branch
     * @param attempt    the attempt number
     * @return {@code true} if all affected pull request were updated correctly. {@code false} if any failed.
     */
    private RepositoryRescopeResult attemptUpdatePullRequestsFor(Repository repository,
                                                                 Map<String, InternalApplicationUser> branchesToUsers,
                                                                 final int attempt) {
        //Find any pull requests which are in the updated repository and, if branch IDs were provided, which relate
        //to the updated branches. To reduce overhead, since these must be kept in memory while rescoping is processed,
        //each pull request is minimized down to its refs, IDs and version.
        //Locked pull requests are filtered out but should trigger another rescope attempt
        int lockedCount = 0;
        ImmutableSet.Builder<MinimalPullRequest> builder = ImmutableSet.builder();
        for (PullRequest pullRequest : findAffectedPullRequests(repository, branchesToUsers.keySet())) {
            if (pullRequest.isLocked()) {
                ++lockedCount;
            } else {
                builder.add(MINIMIZER.apply(pullRequest));
            }
        }

        Set<MinimalPullRequest> pullRequests = builder.build();

        if (pullRequests.isEmpty()) {
            //Don't bother calling the SCM, or doing any other processing, if there are no pull requests available for
            //rescoping
            return new RepositoryRescopeResult(lockedCount, 0);
        }

        //Delegate to the SCM to determine how each affected pull request in the repository should be updated
        SimpleRepositoryRescopeContext context = new SimpleRepositoryRescopeContext(pullRequests);
        scmService.rescope(new RepositoryRescopeCommandParameters.Builder()
                //TODO: Set the refChanges once they are persisted
                .repository(repository)
                .rescopeContext(context)
                .build());

        //Once the SCM has finished processing pull requests, apply the changes it calculated
        int failureCount = 0;
        ListMultimap<ApplicationUser, MinimalPullRequest> usersToPullRequests = groupByUser(pullRequests, branchesToUsers);
        for (ApplicationUser user : usersToPullRequests.keySet()) {
            Set<MinimalPullRequest> userPullRequests = ImmutableSet.copyOf(usersToPullRequests.get(user));
            final Set<SimpleMinimalPullRequest> declined = Sets.intersection(context.declined, userPullRequests);
            final Set<SimpleMinimalPullRequest> merged = Sets.intersection(context.merged, userPullRequests);
            final Set<SimpleMinimalPullRequest> updated = Sets.intersection(context.updated, userPullRequests);
            //Impersonate user that pushed, for creating activities with the right user
            failureCount += securityService.impersonating(user, "Rescoping pull requests for user")
                    .call(new UncheckedOperation<Integer>() {
                        @Override
                        public Integer perform() {
                            return applyUpdate(declined, decline, attempt) +
                                    applyUpdate(merged, merge, attempt) +
                                    applyUpdate(updated, rescope, attempt);
                        }
                    });
        }

        return new RepositoryRescopeResult(lockedCount, failureCount);
    }

    /**
     * Returns all open pull requests that are INCOMING or OUTGOING to the provided repository. The returned Iterable
     * fetches a page of pull requests at a time.
     *
     * @param repository the repository
     * @param branchIds the branches. If it contains {@code null}, all branches are checked.
     * @return all open pull requests that are INCOMING or OUTGOING to the provided repository.
     */
    private Iterable<PullRequest> findAffectedPullRequests(final Repository repository, final Set<String> branchIds) {
        if (branchIds.contains(null)) {
            return searchPullRequests(repository, Collections.<String>emptyList());
        }

        Iterable<PullRequest> result = Collections.emptyList();
        for (List<String> partition : Iterables.partition(branchIds, PullRequestSearchRequest.MAXIMUM_BRANCHES_TO_SEARCH)) {
            Iterable<PullRequest> pullRequests = searchPullRequests(repository, partition);
            result = Iterables.concat(result, pullRequests);
        }
        return result;
    }

    private Map<String, InternalApplicationUser> mapBranchesToUsers(List<InternalRescopeRequest> requests) {
        Map<String, InternalApplicationUser> branchToUser = new HashMap<>();
        for (InternalRescopeRequest request : requests) {
            if (request.getChanges().isEmpty()) {
                branchToUser.put(null, request.getUser());
            } else {
                for (RefChange change : request.getChanges()) {
                    branchToUser.put(change.getRef().getId(), request.getUser());
                }
            }
        }
        return branchToUser;
    }

    private Iterable<PullRequest> searchPullRequests(final Repository repository, final Iterable<String> branchIds) {
        PageRequest initialPageRequest = PageUtils.newRequest(0, 100);

        // In the case where branchIds is empty, intra-repository pull requests are returned twice.
        // That's why the doubled-up pull requests are deduped by the Set around their minimized versions later.
        return Iterables.concat(
                new PagedIterable<>(new PageProvider<PullRequest>() {
                    @Override
                    public Page<PullRequest> get(PageRequest pageRequest) {
                        return pullRequestService.search(new PullRequestSearchRequest.Builder()
                                .toRefIds(branchIds)
                                .toRepositoryId(repository.getId())
                                .state(PullRequestState.OPEN)
                                .withProperties(false)
                                .build(), pageRequest);
                    }
                }, initialPageRequest),
                new PagedIterable<>(new PageProvider<PullRequest>() {
                    @Override
                    public Page<PullRequest> get(PageRequest pageRequest) {
                        return pullRequestService.search(new PullRequestSearchRequest.Builder()
                                .fromRefIds(branchIds)
                                .fromRepositoryId(repository.getId())
                                .state(PullRequestState.OPEN)
                                .withProperties(false)
                                .build(), pageRequest);
                    }
                }, initialPageRequest));
    }

    private static ListMultimap<ApplicationUser, MinimalPullRequest> groupByUser(Set<MinimalPullRequest> pullRequests,
                                                                                 final Map<String, InternalApplicationUser> branchesToUsers) {
        return Multimaps.index(pullRequests, new Function<MinimalPullRequest, ApplicationUser>() {
            @Override
            public ApplicationUser apply(MinimalPullRequest pullRequest) {
                ApplicationUser user = branchesToUsers.get(pullRequest.getFromRef().getId());
                if (user == null) {
                    user = branchesToUsers.get(pullRequest.getToRef().getId());
                    if (user == null) {
                        user = branchesToUsers.get(null);
                        if (user == null) {
                            user = branchesToUsers.values().iterator().next();
                        }
                    }
                }
                return user;
            }
        });
    }

    private static int applyUpdate(Set<SimpleMinimalPullRequest> pullRequests, RescopeUpdateStrategy strategy, int attempt) {
        int failures = 0;

        for (SimpleMinimalPullRequest pullRequest : pullRequests) {
            try {
                strategy.apply(pullRequest);
            } catch (RuntimeException e) {
                ++failures;

                logFailure(String.format(strategy.getErrorFormat(), pullRequest), e, attempt);
            }
        }

        return failures;
    }

    private static void logFailure(String message, Throwable throwable, int attempt) {
        if (attempt < MAX_RESCOPE_ATTEMPTS) {
            log.debug(message, throwable);
        } else {
            log.warn(message, throwable);
        }
    }

    static class MarkerRefChange implements RefChange {
        private final MinimalRef ref;

        public MarkerRefChange(String branchId) {
            this.ref = new SimpleMinimalRef.Builder()
                    .type(StandardRefType.BRANCH)
                    .id(branchId)
                    .displayId(branchId)
                    .build();
        }

        @Nonnull
        @Override
        public String getFromHash() {
            return null;
        }

        @Nonnull
        @Override
        public MinimalRef getRef() {
            return ref;
        }

        @Nonnull
        @Override
        public String getRefId() {
            return getRef().getId();
        }

        @Nonnull
        @Override
        public String getToHash() {
            return null;
        }

        @Nonnull
        @Override
        public RefChangeType getType() {
            return RefChangeType.UPDATE;
        }
    }

    private interface RescopeUpdateStrategy {

        void apply(SimpleMinimalPullRequest pullRequest);

        String getErrorFormat();
    }

    /**
     * Utility class that holds all the state that we need to keep track of during the rescoping of a pull request.
     * This includes:
     * <ul>
     *     <li>A cache for resolving refs</li>
     *     <li>Which pull requests should be declined</li>
     *     <li>Which pull requests should be declined as merged</li>
     *     <li>Which pull requests should be rescoped</li>
     * </ul>
     */
    private static class SimpleRepositoryRescopeContext implements RepositoryRescopeContext {

        private final Set<SimpleMinimalPullRequest> declined;
        private final Set<SimpleMinimalPullRequest> merged;
        private final Set<MinimalPullRequest> pullRequests;
        private final Set<SimpleMinimalPullRequest> updated;

        public SimpleRepositoryRescopeContext(Set<MinimalPullRequest> pullRequests) {
            this.pullRequests = pullRequests;

            declined = Sets.newHashSet();
            merged = Sets.newHashSet();
            updated = Sets.newHashSet();
        }

        @Override
        public void decline(@Nonnull MinimalPullRequest pullRequest) {
            declined.add(checkPullRequest(pullRequest));
        }

        @Override
        public Iterator<MinimalPullRequest> iterator() {
            return pullRequests.iterator();
        }

        @Override
        public void merge(@Nonnull MinimalPullRequest pullRequest) {
            merged.add(checkPullRequest(pullRequest));
        }

        @Override
        public void update(@Nonnull MinimalPullRequest pullRequest, @Nonnull String newFromHash, @Nonnull String newToHash) {
            SimpleMinimalPullRequest minimal = checkPullRequest(pullRequest);
            minimal.newFromHash = newFromHash;
            minimal.newToHash = newToHash;

            updated.add(minimal);
        }

        private static SimpleMinimalPullRequest checkPullRequest(MinimalPullRequest pullRequest) {
            checkNotNull(pullRequest, "pullRequest");
            checkArgument(pullRequest instanceof SimpleMinimalPullRequest,
                    "The provided MinimalPullRequest does not belong to this RepositoryRescopeContext");

            return (SimpleMinimalPullRequest) pullRequest;
        }
    }

    private static class SimpleMinimalPullRequest implements MinimalPullRequest {

        private final PullRequestRef fromRef;
        private final long globalId;
        private final long id;
        private final PullRequestRef toRef;
        private final int version;

        private String newFromHash;
        private String newToHash;

        private SimpleMinimalPullRequest(PullRequest pullRequest) {
            fromRef = pullRequest.getFromRef();
            globalId = InternalConverter.convertToInternalPullRequest(pullRequest).getGlobalId();
            id = pullRequest.getId();
            toRef = pullRequest.getToRef();
            version = pullRequest.getVersion();
        }

        @Override
        public boolean equals(Object o) {
            return o == this ||
                    o instanceof SimpleMinimalPullRequest && globalId == ((SimpleMinimalPullRequest) o).globalId;
        }

        @Nonnull
        @Override
        public PullRequestRef getFromRef() {
            return fromRef;
        }

        @Override
        public long getId() {
            return id;
        }

        @Nonnull
        @Override
        public PullRequestRef getToRef() {
            return toRef;
        }

        @Override
        public int getVersion() {
            return version;
        }

        @Override
        public int hashCode() {
            return Objects.hashCode(globalId);
        }

        @Override
        public String toString() {
            return toRef.getRepository().getProject().getKey() + "/" + toRef.getRepository().getSlug() + "#" + id;
        }
    }
}
