package com.atlassian.stash.internal.pull.comment.drift;

import com.atlassian.bitbucket.comment.DiffCommentAnchor;
import com.atlassian.bitbucket.commit.Commit;
import com.atlassian.bitbucket.commit.MinimalCommit;
import com.atlassian.bitbucket.concurrent.LockService;
import com.atlassian.bitbucket.concurrent.PullRequestLock;
import com.atlassian.bitbucket.event.pull.PullRequestRescopedEvent;
import com.atlassian.bitbucket.permission.Permission;
import com.atlassian.bitbucket.pull.PullRequest;
import com.atlassian.bitbucket.pull.PullRequestCommentAnchorDiffType;
import com.atlassian.bitbucket.pull.RescopeDetails;
import com.atlassian.bitbucket.scm.ScmService;
import com.atlassian.bitbucket.scm.pull.PullRequestEffectiveDiff;
import com.atlassian.bitbucket.user.SecurityService;
import com.atlassian.bitbucket.util.Operation;
import com.atlassian.bitbucket.util.Timer;
import com.atlassian.bitbucket.util.TimerUtils;
import com.atlassian.bitbucket.util.UncheckedOperation;
import com.atlassian.event.api.EventListener;
import com.atlassian.plugin.event.events.PluginFrameworkStartedEvent;
import com.atlassian.stash.internal.pull.InternalPullRequest;
import com.atlassian.stash.internal.pull.InternalPullRequestDiffCommentAnchor;
import com.atlassian.stash.internal.pull.PullRequestAnchorSearchCriteria;
import com.atlassian.stash.internal.pull.comment.CommentUpdateProcessor;
import com.atlassian.stash.internal.pull.comment.InternalPullRequestCommentHelper;
import com.atlassian.stash.internal.repository.InternalRepository;
import com.atlassian.stash.internal.spring.SpringTransactionUtils;
import com.google.common.collect.Maps;
import org.hibernate.HibernateException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DataAccessException;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionException;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallbackWithoutResult;
import org.springframework.transaction.support.TransactionTemplate;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ExecutorService;
import java.util.stream.Stream;

import static com.atlassian.bitbucket.util.MoreCollectors.toImmutableList;
import static com.atlassian.stash.internal.InternalConverter.convertToInternalPullRequest;

/**
 * Applies an injected set of {@link CommentDriftStrategy drift strategies} to update comments when their associated
 * pull request is updated.
 * <p>
 * To constrain the overall load on the system and prevent delay in processing rescopes for other pull requests, the
 * work of calculating comment drift is performed using a separate executor. A new {@link CommentDriftBootstrapper
 * bootstrap} task is scheduled when {@link #process(InternalPullRequest, String, String) processing is requested}.
 * The bootstrapper, when invoked, acquires a lock on the pull request and then delegates the drift processing to a
 * separate {@link CommentDriftOperation operation}. The operation consolidates any pending
 * {@link InternalDriftRequest drift} calculations for the pull request into a single
 * {@link CommentDriftCalculator calculation} and applies it. This achieves two goals:
 * <ul>
 *     <li>Performing comment drift processing on a separately-configurable thread pool
 *     <ul>
 *         <li>Without this, comment drift would be processed synchronously during pull request rescoping</li>
 *         <li>That, in turn, would greatly increase the duration for which a repository-level lock is held,
 *         preventing merging pull requests as well as delaying any further rescopes caused by additional pushes</li>
 *     </ul>
 *     </li>
 *     <li>Ensuring comment drift is never performed concurrently for the same pull request
 *     <ul>
 *         <li>A {@link PullRequestLock pull request-level lock} acquired by the
 *         {@link CommentDriftBootstrapper bootstrapper} ensures only one executor thread is calculating drift for
 *         any given pull request</li>
 *     </ul>
 *     </li>
 * </ul>
 * Drift requests are persisted to the database to ensure that drift requests are performed, even when the server
 * is shutdown halfway through a drift calculation as well as ensuring drift calculation is safe within a clustered
 * environment.
 * <p>
 * When the processing of the drift request is complete, the request is deleted from database in the <em>same</em>
 * transaction being used for the drift calculations. This ensures that if either the updates to the comment anchors
 * or the drift requests trigger a rollback, both are rolled back.
 *
 * @since 1.3
 */
public class DriftCommentUpdateProcessor implements CommentUpdateProcessor {

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

    private final InternalPullRequestCommentHelper commentHelper;
    private final CommentDriftRequestDao driftRequestDao;
    private final ExecutorService executorService;
    private final LockService lockService;
    private final ScmService scmService;
    private final SecurityService securityService;
    private final CommentDriftStrategyChain strategyChain;
    private final TransactionTemplate transactionTemplate;

    private int maxAttempts;

    public DriftCommentUpdateProcessor(InternalPullRequestCommentHelper commentHelper,
                                       CommentDriftRequestDao driftRequestDao, ExecutorService executorService,
                                       LockService lockService, ScmService scmService,
                                       SecurityService securityService, CommentDriftStrategyChain strategyChain,
                                       PlatformTransactionManager transactionManager) {
        this.commentHelper = commentHelper;
        this.driftRequestDao = driftRequestDao;
        this.executorService = executorService;
        this.lockService = lockService;
        this.scmService = scmService;
        this.securityService = securityService;
        this.strategyChain = strategyChain;

        maxAttempts = 5;
        transactionTemplate = new TransactionTemplate(transactionManager, SpringTransactionUtils.REQUIRES_NEW);
    }

    /**
     * Performs any pending comment drift calculation for the provided pull request on the calling thread.
     * Drift calculations are guarded by the {@link LockService} in order to achieve coherence
     *
     * @param pullRequest the pull request
     */
    @Override
    public void maybeProcess(InternalPullRequest pullRequest) {
        new CommentDriftBootstrapper(pullRequest).run();
    }

    @EventListener
    public void onPluginFrameworkStarted(PluginFrameworkStartedEvent ignored) {
        securityService.withPermission(Permission.REPO_WRITE, "Reschedule pending comment drift requests")
                .call(new CommentDriftLoader());
    }

    @EventListener
    public void onPullRequestRescoped(PullRequestRescopedEvent rescopedEvent) {
        transactionTemplate.execute(status -> {
            InternalPullRequest pullRequest = convertToInternalPullRequest(rescopedEvent.getPullRequest());
            try (Timer ignored = TimerUtils.start("Drift: Orphan removed commits activity - " + pullRequest.getGlobalId())) {
                List<InternalPullRequestDiffCommentAnchor> anchors =
                        commentHelper.findDiffAnchors(new PullRequestAnchorSearchCriteria.Builder(pullRequest)
                                .diffType(PullRequestCommentAnchorDiffType.COMMIT)
                                .build(), false);

                if (!anchors.isEmpty()) {
                    RescopeDetails rescopeDetails = rescopedEvent.getRemovedCommits();
                    if (rescopeDetails == null || rescopeDetails.getTotal() > rescopeDetails.getCommits().size()) {
                        //The rescoped event carries no details when the system is unable to determine the outcome
                        //of the rescope. Also, for performance reasons, the rescope details may only have a subset
                        //of the commits which have been removed from the pull request.
                        //Resolve which anchors need to be orphaned by running a rev-list command
                        orphan(findOrphanedAnchors(pullRequest, anchors).stream());
                    } else if (rescopeDetails.getTotal() > 0) {
                        //The rescope details have a complete list of removed commits. Just go through those
                        orphan(anchors.stream().filter(anchor -> rescopeDetails.getCommits().stream()
                                        .anyMatch(commit -> anchorMatches(anchor, commit))));
                    }
                }
            }

            return null;
        });
    }

    /**
     * Inserts a new {@link InternalDriftRequest request} for the provided {@link InternalPullRequest pull request},
     * then submits a {@link CommentDriftBootstrapper bootstrap} task to perform comment drift.
     * <p>
     * In cases where the executor service has no free threads, drift processing may be queued for an indeterminate
     * length of time. During this delay, pull requests pending drift may be rescoped additional times, with each
     * adding its own {@link InternalDriftRequest request} to the list. When the
     * {@link CommentDriftBootstrapper bootstrap} task executes, the drift operation to perform will collapse any number
     * of pending drift requests into a single calculation and perform drift in a single pass. If other bootstrap tasks
     * are pending for the same pull request, when they run there may not be any pending drift requests left. In such
     * cases, they are no-op.
     *
     * @param pullRequest      the pull request which has been updated
     * @param previousFromHash the previous from hash for the pull request
     * @param previousToHash   the previous to hash for the pull request
     */
    @Override
    public void process(InternalPullRequest pullRequest, String previousFromHash, String previousToHash) {
        InternalDriftRequest request = new InternalDriftRequest(
                null, pullRequest, previousFromHash, previousToHash,
                pullRequest.getFromRef().getLatestCommit(), pullRequest.getToRef().getLatestCommit());

        try {
            transactionTemplate.execute(status -> driftRequestDao.create(request));
            executorService.submit(new CommentDriftBootstrapper(request.getPullRequest()));
        } catch (Exception e) {
            log.error("{}:{}@{}: Problem persisting drift request",
                    pullRequest.getScopeRepository().getId(), pullRequest.getId(), pullRequest.getVersion(), e);
        }
    }

    public void setMaxAttempts(int maxAttempts) {
        this.maxAttempts = maxAttempts;
    }

    private boolean anchorMatches(InternalPullRequestDiffCommentAnchor anchor, Commit commit) {
        return Objects.equals(anchor.getToHash(), commit.getId()) && parentMatches(commit, anchor.getFromHash());
    }

    private List<InternalPullRequestDiffCommentAnchor> findOrphanedAnchors(PullRequest pullRequest,
                                                                           List<InternalPullRequestDiffCommentAnchor> currentAnchors) {
        List<InternalPullRequestDiffCommentAnchor> maybeOrphaned = new ArrayList<>(currentAnchors);
        scmService.getPullRequestCommandFactory(pullRequest)
                .commits(commit -> {
                    maybeOrphaned.removeIf(anchor -> anchorMatches(anchor, commit));
                    //If we've found commits associated with all of the anchors, stop streaming commits
                    return !maybeOrphaned.isEmpty();
                }).call();

        return maybeOrphaned;
    }

    private void orphan(Stream<InternalPullRequestDiffCommentAnchor> toOrphan) {
        commentHelper.updateAnchors(toOrphan.map(anchor -> new InternalPullRequestDiffCommentAnchor.Builder(anchor)
                .orphaned(true)
                .build()).collect(toImmutableList()));
    }

    private boolean parentMatches(Commit commit, String hash) {
        return commit.getParents().isEmpty() && hash == null || commit.getParents().stream()
                .map(MinimalCommit::getId)
                .anyMatch(id -> Objects.equals(id, hash));
    }

    /**
     * Bootstraps comment drift processing for a {@link InternalPullRequest pull request} provided during construction.
     * <p>
     * The purpose of this bootstrap task is to {@link PullRequestLock#withLock(
     * PullRequest, Operation) acquire a lock} on the pull request.
     * It then defers processing for any queued {@link InternalDriftRequest requests} to {@link CommentDriftOperation}.
     */
    private class CommentDriftBootstrapper implements Runnable, UncheckedOperation<Void> {

        /**
         * <b>Warning</b>: This pull request <i<may not reflect the current state in the database</i>. It is the state
         * that was available when the bootstrapper was scheduled with the {@code ExecutorService}.
         */
        private final InternalPullRequest cachedPullRequest;

        private CommentDriftBootstrapper(InternalPullRequest cachedPullRequest) {
            this.cachedPullRequest = cachedPullRequest;
        }

        @Override
        public Void perform() {
            return lockService.getPullRequestLock(CommentUpdateProcessor.class.getName())
                    .withLock(cachedPullRequest, new CommentDriftOperation(cachedPullRequest));
        }

        @Override
        public void run() {
            securityService.withPermission(Permission.REPO_READ, "Performing comment drift")
                    .call(this);
        }
    }

    /**
     * Applies various {@link CommentDriftStrategy strategies} to calculate how a pull request's comments have drifted
     * between its {@code previousFromHash}/{@code previousToHash} and its current hashes.
     */
    private class CommentDriftCalculator {

        /**
         * <b>Warning</b>: This pull request <i<may not reflect the current state in the database</i>. It is the state
         * that was available when drift was {@link CommentDriftBootstrapper bootstrapped}.
         */
        private final InternalPullRequest cachedPullRequest;
        private final String previousFromHash;
        private final String previousToHash;
        private final InternalRepository repository;

        private CommentDriftCalculator(InternalPullRequest cachedPullRequest, String previousFromHash,
                                       String previousToHash) {
            this.cachedPullRequest = cachedPullRequest;
            this.previousFromHash = previousFromHash;
            this.previousToHash = previousToHash;

            repository = cachedPullRequest.getScopeRepository();
        }

        /**
         * Retrieves all non-{@link InternalPullRequestDiffCommentAnchor#isOrphaned() orphaned} anchors for the
         * rescoped pull request and applies the injected {@link CommentDriftStrategy drift strategies} to move
         * the anchors according to how path contents have changed.
         */
        @SuppressWarnings("ConstantConditions")
        public void calculate(List<InternalDriftRequest> driftRequests) {
            PullRequestEffectiveDiff effectiveDiff;
            Map<PullRequestEffectiveDiff, List<InternalPullRequestDiffCommentAnchor>> previousDiffs;

            try (Timer ignored = TimerUtils.start("Drift: Map anchors " + cachedPullRequest.getGlobalId())) {
                List<InternalPullRequestDiffCommentAnchor> anchors =
                        commentHelper.findDiffAnchors(new PullRequestAnchorSearchCriteria.Builder(cachedPullRequest)
                                .diffType(PullRequestCommentAnchorDiffType.EFFECTIVE)
                                .build(), false);
                if (anchors.isEmpty()) {
                    log.debug("{}:{}@{}: No active diff comments",
                            repository.getId(), cachedPullRequest.getId(), cachedPullRequest.getVersion());
                    return;
                }

                //Retrieving the effective diff _ignores_ the provided PullRequest and always retrieves the current
                //state from the database. That means the hashes on the cachedPullRequest don't matter here
                effectiveDiff = scmService.getPullRequestCommandFactory(cachedPullRequest)
                        .effectiveDiff()
                        .call();
                previousDiffs = mapPreviousDiffs(anchors, effectiveDiff);
            }

            if (previousDiffs.isEmpty()) {
                StringBuilder builder = new StringBuilder()
                        .append(repository.getId())
                        .append(":")
                        .append(cachedPullRequest.getId())
                        .append("@")
                        .append(cachedPullRequest.getVersion())
                        .append(": Comments have already been drifted (")
                        .append(effectiveDiff.getSinceId())
                        .append(", ")
                        .append(effectiveDiff.getUntilId())
                        .append(") when processing ")
                        .append(driftRequests.size())
                        .append(" drift request(s):");
                for (InternalDriftRequest driftRequest : driftRequests) {
                    builder.append("\n\t").append(driftRequest);
                }
                log.info(builder.toString());

                return;
            } else if (previousDiffs.size() > 1) {
                log.warn("{}:{}@{}: Non-orphaned comments span {} diffs", repository.getId(),
                        cachedPullRequest.getId(), cachedPullRequest.getVersion(), previousDiffs.size());
            }

            for (Map.Entry<PullRequestEffectiveDiff, List<InternalPullRequestDiffCommentAnchor>> entry : previousDiffs.entrySet()) {
                PullRequestEffectiveDiff previousDiff = entry.getKey();
                DriftContext context = new DriftContext(repository, cachedPullRequest, previousFromHash,
                        previousToHash, effectiveDiff, previousDiff, entry.getValue());

                try (Timer ignored = TimerUtils.start("Drift: " + strategyChain + " " + cachedPullRequest.getGlobalId())) {
                    strategyChain.applyTo(context);
                }

                List<InternalPullRequestDiffCommentAnchor> driftedAnchors = context.done();
                try (Timer ignored = TimerUtils.start("Drift: Update anchors " + cachedPullRequest.getGlobalId())) {
                    commentHelper.updateAnchors(driftedAnchors);
                }

                log.debug("{}:{}@{}: Successfully drifted {} anchors from ({}, {}) to ({}, {})",
                        repository.getId(), cachedPullRequest.getId(), cachedPullRequest.getVersion(),
                        driftedAnchors.size(), previousDiff.getSinceId(), previousDiff.getUntilId(),
                        effectiveDiff.getSinceId(), effectiveDiff.getUntilId());
            }
        }

        /**
         * Infers the previous {@link PullRequestEffectiveDiff effective diffs} by examining the
         * {@link InternalPullRequestDiffCommentAnchor#getFromHash() from hash} and
         * {@link InternalPullRequestDiffCommentAnchor#getToHash() to hash} for all of the anchors
         * in the provided list.
         * <p>
         * This <i>should</i> always return a {@code Map} with a single entry. However, due to how
         * new comments are anchored, it's possible the returned map may be empty or contain more
         * than one diff. Drift must be calculated separately for each diff.
         *
         * @param anchors the anchors from which the previous effective diffs should be inferred
         * @return The inferred effective diffs
         */
        @SuppressWarnings("ConstantConditions")
        private Map<PullRequestEffectiveDiff, List<InternalPullRequestDiffCommentAnchor>> mapPreviousDiffs(
                List<InternalPullRequestDiffCommentAnchor> anchors, PullRequestEffectiveDiff currentDiff) {
            Map<PullRequestEffectiveDiff, List<InternalPullRequestDiffCommentAnchor>> previousDiffs = Maps.newHashMap();
            for (DiffCommentAnchor anchor : anchors) {
                PullRequestEffectiveDiff diff = new PullRequestEffectiveDiff(anchor.getToHash(), anchor.getFromHash());
                if (currentDiff.equals(diff)) {
                    //Ignore any comment which is in the current diff already. There are two primary possibilities for
                    //how we got here:
                    //- Just-in-time drift (performed when a user views a pull request) racing with background drift
                    //  - This is completely harmless; the comments have already been properly drifted by one or the
                    //    other and so the one that runs second has nothing to do
                    //  - This case becomes even more likely in a clustered environment, where multiple nodes are all
                    //    capable of processing drift
                    //- A race condition between pull request rescoping, newly-added comments and comment drift
                    //  - User is viewing diff X, between commits A and B
                    //  - The server rescopes the pull request to commits A and C
                    //  - User adds a comment (still viewing the outdated diff)
                    //  - The system records the comment against A and C, because that's current, even though the user
                    //    was not viewing that diff
                    //  - Comment drift runs and views the comment as already up-to-date, arriving here
                    //
                    //The first case is significantly more common than the second, but it's not readily possible, here,
                    //to tell the two apart. This is logged at info level instead of warn because it usually doesn't
                    //mean anything. Info level still appears in the logs, however, so the message is still present in
                    //case it turns out to be the race condition.
                    log.info("{}:{}@{}: Comment {} is already in the current diff ({}, {}) and will not be drifted",
                            repository.getId(), cachedPullRequest.getId(), cachedPullRequest.getVersion(),
                            anchor.getComment().getId(), currentDiff.getSinceId(), currentDiff.getUntilId());
                    continue;
                }

                previousDiffs.computeIfAbsent(diff, key -> new ArrayList<>())
                        .add((InternalPullRequestDiffCommentAnchor) anchor);
            }

            return previousDiffs;
        }
    }

    /**
     * Loads any pending {@link InternalDriftRequest drift requests} and reschedules them to allow the background
     * threads to continue processing them.
     *
     * @since 2.9.5
     */
    private class CommentDriftLoader extends TransactionCallbackWithoutResult implements UncheckedOperation<Void> {

        /**
         * Loads pull requests which have at least one pending drift request which has attempts remaining and
         * schedules them for processing.
         *
         * @param status ignored
         */
        @Override
        protected void doInTransactionWithoutResult(TransactionStatus status) {
            List<InternalPullRequest> pullRequests = driftRequestDao.findPendingPullRequests(maxAttempts);
            for (InternalPullRequest pullRequest : pullRequests) {
                executorService.submit(new CommentDriftBootstrapper(pullRequest));
            }
        }

        /**
         * Provides the {@link UncheckedOperation} for use with the {@link SecurityService} and then delegates to
         * the {@code TransactionTemplate} to load pending drift.
         *
         * @return {@code null}
         */
        @Override
        public Void perform() {
            transactionTemplate.execute(this);

            return null;
        }
    }

    /**
     * Retrieve the pending {@link InternalDriftRequest requests} from the database for the pull request provided
     * during construction, consolidates the hashes if multiple requests are pending, and then calculates drift.
     * <p>
     * If the list of requests is empty, it is assumed that another operation on another thread has already performed
     * comment drift for the pull request and no further work is done. Otherwise, if the list contains multiple
     * {@link InternalDriftRequest requests}, the hashes of the <i>first</i> request are paired with the
     * {@link InternalPullRequest pull request} of the <i>last</i> request and comments are drifted for all versions
     * in a single pass.
     * <p>
     * Note: This operation assumes that the pull request is locked, guaranteeing that only a single drift operation
     * is being performed at a time. Failure to satisfy this assumption will lead to unexpected, and likely unpleasant,
     * results.
     */
    private class CommentDriftOperation implements UncheckedOperation<Void> {

        /**
         * <b>Warning</b>: This pull request <i<may not reflect the current state in the database</i>. It is the state
         * that was available when drift was {@link CommentDriftBootstrapper bootstrapped}.
         */
        private final InternalPullRequest cachedPullRequest;

        private CommentDriftOperation(InternalPullRequest cachedPullRequest) {
            this.cachedPullRequest = cachedPullRequest;
        }

        @Override
        public Void perform() {
            // Retrieve the current drift requests. These will remain stored in
            // the database until they are successfully performed
            List<InternalDriftRequest> drifts = getPendingDrifts(cachedPullRequest);
            if (drifts.isEmpty()) {
                log.debug("{}:{}@{}: No rescopes are pending drift", cachedPullRequest.getScopeRepository().getId(),
                        cachedPullRequest.getId(), cachedPullRequest.getVersion());
                return null;
            }

            //Perform the drift calculation and update the drift requests in database in the same transaction to
            //guarantee a consistent state in the database.
            CommentDriftCalculator calculator = calculatorFor(drifts);
            try {
                transactionTemplate.execute(status -> {
                    try (Timer ignored = TimerUtils.start("Drift: Calculate for " + calculator)) {
                        calculator.calculate(drifts);
                        driftRequestDao.deleteAll(drifts);
                    }
                    return null;
                });
            } catch (RuntimeException e) {
                maybeReschedule(drifts, e);
            }

            return null;
        }

        private CommentDriftCalculator calculatorFor(List<InternalDriftRequest> drifts) {
            if (drifts.size() > 1) {
                log.debug("{}:{}@{}: Calculating combined drift for {} rescopes",
                        cachedPullRequest.getScopeRepository().getId(), cachedPullRequest.getId(),
                        cachedPullRequest.getVersion(), drifts.size());
            }
            InternalDriftRequest first = drifts.get(0);

            return new CommentDriftCalculator(cachedPullRequest, first.getOldFromHash(), first.getOldToHash());
        }

        private List<InternalDriftRequest> getPendingDrifts(InternalPullRequest pullRequest) {
            return transactionTemplate.execute(status -> driftRequestDao.findByGlobalId(pullRequest.getGlobalId()));
        }

        private boolean isRecoverable(Exception e) {
            return DataAccessException.class.isAssignableFrom(e.getClass()) ||
                    TransactionException.class.isAssignableFrom(e.getClass()) ||
                    HibernateException.class.isAssignableFrom(e.getClass());
        }

        /**
         * A count of failed attempts is maintained on the newest pending drift request.  If the number of
         * failed attempts is less than the configured max and the error is considered "recoverable" (e.g., a
         * concurrent modification exception from the database), then drift for the affected pull request is
         * retried.  Otherwise an error is logged and drift for the affected pull request is not retried
         * further, until one of the following happens:
         * <ul>
         *     <li>Another rescope happens and a newer request is created (with a 0 failed count), or</li>
         *     <li>The system is restarted</li>
         * </ul>
         *
         * @param drifts the drift requests being processed
         * @param e      the exception which terminated processing
         */
        private void maybeReschedule(List<InternalDriftRequest> drifts, RuntimeException e) {
            InternalDriftRequest newestRequest = drifts.get(drifts.size() - 1);
            int failedAttempts = newestRequest.getAttempts() + 1;
            boolean recoverable = isRecoverable(e);

            if (recoverable && failedAttempts <= maxAttempts) {
                log.info("{}:{}@{}: Failed to drift comments (Attempt {} of {}). Rescheduling",
                        cachedPullRequest.getScopeRepository().getId(), cachedPullRequest.getId(),
                        cachedPullRequest.getVersion(), failedAttempts, maxAttempts, e);

                try {
                    // Drift failed, but may be recoverable. Update the number of attempts we've made, and
                    // reschedule the pull request for drift processing
                    updateAttempts(drifts, failedAttempts);
                    executorService.submit(new CommentDriftBootstrapper(cachedPullRequest));
                } catch (RuntimeException ex) {
                    log.warn("{}:{}@{}: Comment drift could not be rescheduled. No further attempts will be made",
                            cachedPullRequest.getScopeRepository().getId(), cachedPullRequest.getId(),
                            cachedPullRequest.getVersion(), ex);
                }
            } else {
                log.error("{}:{}@{}: {} calculating comment drift ({} attempts).\n\tFirst: {}\n\tLast: {}",
                        cachedPullRequest.getScopeRepository().getId(), cachedPullRequest.getId(),
                        cachedPullRequest.getVersion(), recoverable ? "Error" : "Unrecoverable error",
                        failedAttempts, drifts.get(0), newestRequest, e);

                try {
                    // Drift processing hit an unrecoverable error. These requests should not be retried. However,
                    // they cannot be removed from the database. Their hashes are important for attempting future
                    // drift processing. To prevent them from being retried, we update them to maxAttempts + 1
                    updateAttempts(drifts, maxAttempts + 1);
                } catch (RuntimeException ex) {
                    log.warn("{}:{}@{}: Aborted drift requests could not be removed from the database. They may be " +
                            "retried after the next system restart", cachedPullRequest.getScopeRepository().getId(),
                            cachedPullRequest.getId(), cachedPullRequest.getVersion(), ex);
                }
            }
        }

        /**
         * Records the number of failed attempts against the <i>oldest</i> {@link InternalDriftRequest request},
         * and deletes any others.
         * <p>
         * {@link #calculatorFor} uses the <i>oldest</i> request to perform comment drift, ensuring comments are
         * drifted from the correct starting point. If there are multiple requests, only the oldest one needs to
         * be retained. To try and ensure the database doesn't build up rows over time, if drift fails repeatedly
         * for a pull request, newer drift requests are removed.
         *
         * @param drifts   the drift requests loaded when processing started
         * @param attempts the number of failed attempts to record against the oldest drift request
         */
        private void updateAttempts(List<InternalDriftRequest> drifts, int attempts) {
            transactionTemplate.execute(new TransactionCallbackWithoutResult() {

                @Override
                protected void doInTransactionWithoutResult(TransactionStatus status) {
                    // Update the attempts count on the oldest drift request, which has the hashes we need for
                    // subsequent drift attempts
                    driftRequestDao.updateAttempts(drifts.get(0), attempts);

                    // If there are other drift requests, delete them from the database. Their hashes and attempt
                    // counts aren't relevant for subsequent attempts
                    if (drifts.size() > 1) {
                        driftRequestDao.deleteAll(drifts.subList(1, drifts.size()));
                    }
                }
            });
        }
    }
}
