package com.atlassian.bitbucket.mesh.git.pull;

import com.atlassian.bitbucket.mesh.util.SortedShaSet;
import com.google.common.collect.ImmutableList;

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

import static com.atlassian.bitbucket.mesh.util.ShaUtils.hashesMatch;
import static java.util.Objects.requireNonNull;

/**
 * Helper class that calculates how one or more {@link PullRequestRescope rescopes} affect their target pull requests.
 * The analyzer expects to receive all commits that are reachable from the oldFrom, oldTo, newFrom and newTo commits,
 * but are not reachable from their common ancestor on the {@link #onCommit(String, List)} method. It expects to
 * receive these commits in topological order: no commit is received before any of its children.
 * <p>
 * The {@link #onCommit(String, List)} method may be called with any other commits; these are ignored.
 */
@NotThreadSafe
public class BatchPullRequestRescopeAnalyzer {

    private final PullRequestRescopeAnalysisContext analysisContext;
    private final List<SinglePullRequestRescopeAnalyzer> analyzers;
    private final List<PullRequestRescopeAnalysisResult> results;

    private BatchPullRequestRescopeAnalyzer(Builder builder) {
        analysisContext = new PullRequestRescopeAnalysisContext(builder.maxCommitIds);
        analyzers = builder.rescopes.stream()
                .map(rescope -> new SinglePullRequestRescopeAnalyzer(rescope, analysisContext))
                .collect(Collectors.toList());
        results = new ArrayList<>();
    }

    /**
     * Signals the end of all analysis and releases as much resources as possible
     */
    public void close() {
        for (SinglePullRequestRescopeAnalyzer analyzer : analyzers) {
            results.add(analyzer.getResult());
        }
        analyzers.clear();
    }

    /**
     * @return list of the {@link PullRequestRescopeAnalysisResult results} of rescope analysis, 1 result for each
     *         {@link Builder#rescope(PullRequestRescope) provided rescope}
     */
    @Nonnull
    public List<PullRequestRescopeAnalysisResult> getResults() {
        if (analyzers.isEmpty()) {
            return Collections.unmodifiableList(results);
        }
        return ImmutableList.<PullRequestRescopeAnalysisResult>builder()
                .addAll(results)
                .addAll(analyzers.stream()
                        .map(SinglePullRequestRescopeAnalyzer::getResult)
                        .collect(Collectors.toList()))
                .build();
    }

    /**
     * @param commitId the commit ID
     * @param parentIds the IDs of the commit's parents
     * @return {@code false} if all rescopes have been analyzed and no more commits are needed. {@code true} if more
     *         commits are needed to complete the analysis
     */
    public boolean onCommit(@Nonnull String commitId, @Nonnull List<String> parentIds) {
        analysisContext.onCommit(commitId, parentIds);
        int remainingAnalyzers = 0;
        Iterator<SinglePullRequestRescopeAnalyzer> it = analyzers.iterator();
        while (it.hasNext()) {
            SinglePullRequestRescopeAnalyzer analyzer = it.next();
            if (analyzer.onCommit(commitId, parentIds)) {
                ++remainingAnalyzers;
            } else {
                // analyzer has figured out what the outcome of the rescope is (decline/merge/update).
                it.remove();
                results.add(analyzer.getResult());
            }
        }
        return remainingAnalyzers > 0;
    }

    public static class Builder {

        private final int maxCommitIds;
        private final List<PullRequestRescope> rescopes;

        public Builder(int maxCommitIds) {
            this.maxCommitIds = maxCommitIds;

            rescopes = new ArrayList<>();
        }

        public BatchPullRequestRescopeAnalyzer build() {
            return new BatchPullRequestRescopeAnalyzer(this);
        }

        @Nonnull
        public Builder rescope(@Nonnull PullRequestRescope rescope) {
            rescopes.add(requireNonNull(rescope, "rescope"));
            return this;
        }
    }

    /**
     * AncestorTracker that maintains a sorted array of heads. Duplicates are not allowed. Two hashes are deemed to be
     * identical when one is a prefix of the other or vice versa.
     * <p>
     * Note: this uses SortedShaSet instead of a TreeSet with a special comparator to avoid the (memory) overhead of
     * the TreeSet and the iterators to loop over it. Eliminating this overhead speeds up the rescope analysis
     * significantly (> 50% in some cases) by massively cutting down the amount of gc required.
     */
    private static class AncestorTracker {

        private int openCount = -1;
        private SortedShaSet heads;
        private boolean updated;

        AncestorTracker(String commitId) {
            heads = new SortedShaSet();
            heads.add(commitId);
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) {
                return true;
            }
            if (o == null || getClass() != o.getClass()) {
                return false;
            }
            AncestorTracker that = (AncestorTracker) o;
            return Objects.equals(heads, that.heads);
        }

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

        void close() {
            if (--openCount == 0) {
                heads = null;
            }
        }

        boolean isOpen() {
            return openCount > 0;
        }

        boolean isUpdated() {
            return updated;
        }

        void onCommit(@Nonnull String commitId, @Nonnull List<String> parentIds) {
            updated = heads.remove(commitId);
            if (updated) {
                for (String parentId : parentIds) {
                    heads.add(parentId);
                }
            }
        }

        AncestorTrackerStub open() {
            openCount = openCount == -1 ? 1 : openCount + 1;
            return new AncestorTrackerStub(this);
        }
    }

    /**
     * Stub that ensures that the {@link AncestorTracker#close} is called exactly once when the stub is closed
     */
    private static class AncestorTrackerStub {

        private AncestorTracker tracker;

        private AncestorTrackerStub(AncestorTracker tracker) {
            this.tracker = tracker;
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) {
                return true;
            }
            if (o == null || getClass() != o.getClass()) {
                return false;
            }
            AncestorTrackerStub that = (AncestorTrackerStub) o;
            return Objects.equals(tracker, that.tracker);
        }

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

        void close() {
            if (tracker != null) {
                tracker.close();
                tracker = null;
            }
        }

        boolean isUpdated() {
            return get().isUpdated();
        }

        private AncestorTracker get() {
            if (tracker == null) {
                throw new IllegalStateException("AncestorTracker has already been closed");
            }
            return tracker;
        }
    }

    private static class PullRequestRescopeAnalysisContext {

        private final int maxCommitIds;
        private final Map<String, AncestorTracker> trackers;

        PullRequestRescopeAnalysisContext(int maxCommitIds) {
            this.maxCommitIds = maxCommitIds;

            trackers = new HashMap<>();
        }

        @Nonnull
        AncestorTrackerStub openTracker(@Nonnull String commitId) {
            AncestorTracker tracker = trackers.computeIfAbsent(commitId, AncestorTracker::new);
            return tracker.open();
        }

        void onCommit(@Nonnull String commitId, @Nonnull List<String> parentIds) {
            Iterator<AncestorTracker> it = trackers.values().iterator();
            while (it.hasNext()) {
                AncestorTracker tracker = it.next();
                if (tracker.isOpen()) {
                    tracker.onCommit(commitId, parentIds);
                } else {
                    it.remove();
                }
            }
        }

        int getMaxCommitIds() {
            return maxCommitIds;
        }
    }

    /**
     * Helper class that calculates how a {@link PullRequestRescope rescope} changes its target pull request.
     */
    @NotThreadSafe
    private static class SinglePullRequestRescopeAnalyzer {

        private final PullRequestRescope rescope;
        private final String searchTarget;

        // -- analysis temporary state
        private boolean anyInNewScope;
        private boolean anyRemoved;
        private boolean fromBranchDeleted;
        private boolean fromCommonAncestorFound;
        private boolean toCommonAncestorFound;
        private AncestorTrackerStub oldFromAncestors;
        private AncestorTrackerStub oldToAncestors;
        private AncestorTrackerStub newFromAncestors;
        private AncestorTrackerStub newToAncestors;

        // -- analysis results
        private PullRequestRescopeAnalysisResult.Builder resultBuilder;
        private PullRequestRescopeAnalysisResult result;

        SinglePullRequestRescopeAnalyzer(PullRequestRescope rescope, PullRequestRescopeAnalysisContext context) {
            this.rescope = rescope;
            this.resultBuilder = new PullRequestRescopeAnalysisResult.Builder(rescope, context.getMaxCommitIds());

            fromBranchDeleted = rescope.getNewFromCommit() == null;
            searchTarget = fromBranchDeleted ? rescope.getOldFromCommit() : rescope.getNewFromCommit();
            if (fromBranchDeleted && hashesMatch(rescope.getOldToCommit(), rescope.getNewToCommit())) {
                // source branch has been deleted, but target branch is unchanged
                setResult(PullRequestRescopeOutcome.DECLINE);
            } else if (rescope.getNewToCommit() == null) {
                // target branch has been deleted
                setResult(PullRequestRescopeOutcome.DECLINE);
            } else {
                oldFromAncestors = context.openTracker(rescope.getOldFromCommit());
                oldToAncestors = context.openTracker(rescope.getOldToCommit());
                newFromAncestors = rescope.getNewFromCommit() == null
                        ? null : context.openTracker(rescope.getNewFromCommit());
                newToAncestors = context.openTracker(rescope.getNewToCommit());

                fromCommonAncestorFound = oldFromAncestors.equals(newFromAncestors);
                toCommonAncestorFound = oldToAncestors.equals(newToAncestors);
            }
        }

        public boolean isDone() {
            return result != null;
        }

        public boolean onCommit(@Nonnull String commitId, @Nonnull List<String> parentIds) {
            if (isDone()) {
                return false;
            }

            boolean ancestorOfNewTo = newToAncestors.isUpdated();
            if (ancestorOfNewTo && matchesAny(searchTarget, parentIds)) {
                // the search target is reachable from the to-branch, so the scope is empty

                // TODO: improve merge hash detection. For now this only detects the case where the searchTarget was
                // merged into the target branch directly. It detects most merges correctly, but here's one case that is
                // not yet supported:
                //
                // a --- b --- c --- f --- g        [to: c -> g]
                //  \               /
                //    -- d --- e --                 [from: d -> deleted]
                // In this case, searchTarget is d which is discovered to be reachable from the to-branch when e is
                // processed, which is not a merge commit.

                // A merge commit was detected that directly merges in searchTarget
                if (hashesMatch(commitId, rescope.getNewToCommit()) && parentIds.size() > 1) {
                    resultBuilder.mergeCommit(commitId);
                }
                handleEmptyScope();
                return false;
            }

            boolean ancestorOfOldTo = oldToAncestors.isUpdated();
            boolean ancestorOfOldFrom = oldFromAncestors.isUpdated();
            boolean ancestorOfNewFrom = !fromBranchDeleted && newFromAncestors.isUpdated();
            boolean inOldScope = ancestorOfOldFrom && !ancestorOfOldTo;
            boolean inNewScope = ancestorOfNewFrom && !ancestorOfNewTo;
            anyInNewScope |= inNewScope;

            if (inOldScope && !inNewScope) {
                anyRemoved = true;
                resultBuilder.removedCommit(commitId);
            } else if (inNewScope && !inOldScope) {
                resultBuilder.addedCommit(commitId);
            }

            if (ancestorOfOldFrom || ancestorOfNewFrom || ancestorOfOldTo || ancestorOfNewTo) {
                if (!fromBranchDeleted && !fromCommonAncestorFound) {
                    fromCommonAncestorFound = oldFromAncestors.equals(newFromAncestors);
                }
                if (!toCommonAncestorFound) {
                    toCommonAncestorFound = oldToAncestors.equals(newToAncestors);
                }

                if (toCommonAncestorFound) {
                    if (fromBranchDeleted && oldFromAncestors.equals(newToAncestors)) {
                        // common ancestor found for [oldFromHash, oldToHash, newToHash]. The from-branch was deleted
                        // but not merged into the to-branch
                        setResult(PullRequestRescopeOutcome.DECLINE);
                    }
                    if (!fromBranchDeleted && fromCommonAncestorFound) {
                        // at this point, we know what commits have been added to and removed from the scope. Let's try
                        // to classify the outcome of the rescope: UPDATE / DECLINE / MERGE
                        if (anyInNewScope || !anyRemoved) {
                            // a commit was found in the new scope, or we have determined for sure that no commits were
                            // _removed_ from the scope ==> there's at least one commit in the new scope -> UPDATE
                            setResult(PullRequestRescopeOutcome.UPDATE);
                        } else {
                            // we're not sure yet whether the new scope is empty or not. We'll have to continue walking
                            // the commit graph until we find a commit that is in the new scope (and hence classify as
                            // 'UPDATE'), or until we find the common ancestor of oldFrom, oldTo, newFrom and newTo and
                            // classify as 'DECLINE' or 'MERGE'
                            if (isCommonAncestorFound()) {
                                // common ancestor for [oldFromHash, newFromHash, oldToHash, newToHash] has been found
                                // without finding any commits in scope.
                                handleEmptyScope();
                            }
                        }
                    }
                }
            }

            return !isDone();
        }

        public PullRequestRescope getRescope() {
            return rescope;
        }

        @Nonnull
        public PullRequestRescopeAnalysisResult getResult() {
            return result != null ? result : resultBuilder.build();
        }

        private static boolean matchesAny(String hash, Iterable<String> hashes) {
            for (String h : hashes) {
                if (hashesMatch(h, hash)) {
                    return true;
                }
            }
            return false;
        }

        private void handleEmptyScope() {
            if (hashesMatch(rescope.getOldToCommit(), rescope.getNewToCommit())) {
                // The scope is empty, but the to-branch hasn't changed. This can only mean that the from-branch has
                // been rewound to an ancestor of the to-branch, thereby stripping all commits from the PR.
                setResult(PullRequestRescopeOutcome.DECLINE);
            } else {
                // to-branch has changed. There are a couple of possibilities here:
                // - from-branch has not been updated:
                //   - the from-branch has been merged into the to-branch  ==> MERGE
                //   - the to-branch has been rebased onto the from-branch ==> MERGE
                // - the from-branch _has_ been updated:
                //   - from-branch updated, then merged into the to-branch ==> MERGE
                //   - from-branch updated, then to-branch was rebased onto the from-branch ==> MERGE
                //   - from-branch was rewound and the to-branch has been updated. This is where it becomes fuzzy:
                //     - from-branch was stripped to empty and an independent update happened to to-branch ==> DECLINE
                //     - from-branch was squash-merged and fast-forward merged into to-branch ==> MERGE
                // Because we cannot easily distinguish between the last two scenarios, we'll pick 'MERGE' in both
                // cases.
                setResult(PullRequestRescopeOutcome.MERGE);
            }
        }

        private boolean isCommonAncestorFound() {
            return newFromAncestors.equals(newToAncestors);
        }

        private void setResult(PullRequestRescopeOutcome outcome) {
            result = resultBuilder.outcome(outcome).build();
            resultBuilder = null;

            // clear the heads collections to allow the memory to be gc-d
            if (oldFromAncestors != null) {
                oldFromAncestors.close();
            }
            if (oldToAncestors != null) {
                oldToAncestors.close();
            }
            if (newFromAncestors != null) {
                newFromAncestors.close();
            }
            if (newToAncestors != null) {
                newToAncestors.close();
            }
            oldFromAncestors = null;
            oldToAncestors = null;
            newFromAncestors = null;
            newToAncestors = null;
        }
    }
}
