package com.atlassian.bitbucket.mesh.transaction;

import com.atlassian.bitbucket.mesh.config.GitCompatibilityConfig;
import com.atlassian.bitbucket.mesh.execution.ExecutionLocal;
import com.atlassian.bitbucket.mesh.execution.ExecutionManager;
import com.atlassian.bitbucket.mesh.git.GitConstants;
import com.atlassian.bitbucket.mesh.git.GitObjectInfo;
import com.atlassian.bitbucket.mesh.git.builder.GitProcessBuilderFactory;
import com.atlassian.bitbucket.mesh.git.builder.catfile.InfoBatchCheckCatFilesStdioHandler;
import com.atlassian.bitbucket.mesh.git.hook.*;
import com.atlassian.bitbucket.mesh.git.hook.handler.AbstractGitHookHandler;
import com.atlassian.bitbucket.mesh.git.ref.RefUpdate;
import com.atlassian.bitbucket.mesh.repository.Repository;
import com.atlassian.bitbucket.mesh.util.Person;
import com.atlassian.bitbucket.mesh.util.ShaUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import java.io.IOException;
import java.time.Duration;
import java.util.*;
import java.util.stream.Collectors;

import static com.atlassian.bitbucket.mesh.git.GitConstants.ENV_BB_GIT_COMMAND;
import static com.atlassian.bitbucket.mesh.transaction.ReferenceTransactionState.*;
import static java.util.stream.Collectors.toList;

/**
 * {@code git reference-transaction} hook handler that parses the hooks stdin, filters out git-internal transaction
 * events and forwards the "real" events to the registered {@link ReferenceTransactionListener listeners}.
 * <p>
 * More specifically, this handler
 * <ul>
 *     <li>filters out ABORTED and COMMITTED callbacks not preceded by a PREPARED callback. These callbacks only
 *     happen as a result of {@code git pack-refs}, which does not lead to publicly visible changes, or when refs
 *     are deleted that are (also) located in {@code packed-refs}. In this latter case, the callback for the
 *     packed-refs backend is suppressed, but the "real" callback is propagated to the listeners.
 *     </li>
 *     <li>filters out ABORTED and COMMITTED callbacks for the packed-refs backend when deleting refs. In this case,
 *     git also triggered possibly more complete "real" callbacks. Only these latter callbacks are propagated to
 *     the listeners.
 *     </li>
 *     <li>resolves the {@link RefUpdate#getOldHash() old hash} on each updated ref. On a normal
 *     {@link RefUpdate}, a {@link ShaUtils#NULL_SHA1} old hash means that the ref was created, but on a
 *     {@code reference-transaction} callback this means that the ref was created, or forcibly updated. To provide
 *     correct {@link RefUpdate} semantics for listeners, the old hash is resolved before being propagated to the
 *     listeners.</li>
 * </ul>
 */
public class GitReferenceTransactionHookHandler extends AbstractGitHookHandler {

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

    private final GitProcessBuilderFactory builderFactory;
    private final GitCompatibilityConfig compatibilityConfig;
    private final ExecutionLocal<ReferenceTransactionContext> transactionContext;
    private final List<ReferenceTransactionListener> transactionListeners;
    private final UpdateRefsContextHolder updateRefsContextHolder;

    public GitReferenceTransactionHookHandler(GitCompatibilityConfig compatibilityConfig,
                                              GitProcessBuilderFactory builderFactory,
                                              ExecutionManager executionManager,
                                              List<ReferenceTransactionListener> transactionListeners,
                                              UpdateRefsContextHolder updateRefsContextHolder) {
        super(GitHook.REFERENCE_TRANSACTION);

        this.builderFactory = builderFactory;
        this.compatibilityConfig = compatibilityConfig;
        this.transactionListeners = transactionListeners;
        this.updateRefsContextHolder = updateRefsContextHolder;

        transactionContext = executionManager.newExecutionLocal(this::maybeClose);
    }

    @Override
    public boolean handle(@Nonnull Repository repository, @Nonnull HookRequest request,
                          @Nonnull GitHookOutputWriter outputWriter) {
        if (log.isTraceEnabled()) {
            log.trace("[{}] reference-transaction {args: {}, stdin: {} bytes}",
                    repository, request.getArguments(), request.getStdin().size());
        }

        List<String> args = request.getArguments();
        if (args.isEmpty()) {
            log.warn("[{}] Rejecting reference transaction because the state is missing", repository);
            return false;
        }

        ReferenceTransactionState state;
        try {
            state = ReferenceTransactionState.valueOf(args.get(0).toUpperCase(Locale.ROOT));
        } catch (IllegalArgumentException e) {
            log.warn("[{}] Rejecting reference transaction because state {} is unknown", request, args.get(0));
            return false;
        }

        //First check whether earlier callbacks were made in the same execution context, and if so whether the
        //current callback is compatible with that context.
        ReferenceTransactionContext context = transactionContext.computeIfAbsent(
                () -> new ReferenceTransactionContext(repository, request.getCommand()));
        //Context will be null if this hook handler is called without an active execution context,
        //which can happen when refs are updated or packed in background tasks and/or unit tests.
        if (context != null && !repository.equals(context.repository)) {
            if (context.publishedState == PREPARED) {
                //This could happen when multiple git operations update refs in different repositories in parallel
                //(barring bugs). We currently do not support that, so reject the update.
                log.error("[{}] Rejecting reference transaction because a conflicting transaction for {} " +
                        "is currently PREPARED", repository, context.repository);
                return false;
            } else {
                //This can happen when refs are updated in multiple repositories from the same RPC call (but
                //different git operations)
                log.debug("[{}] Resetting incompatible {} transaction for repository {}", repository,
                        context.publishedState == null ? "INITIALIZED" : context.publishedState, context.repository);
                context = new ReferenceTransactionContext(repository, request.getEnvironment().get(ENV_BB_GIT_COMMAND));
                transactionContext.set(context);
            }
        }

        try {
            ReferenceTransactionParser parser = new ReferenceTransactionParser(
                    context, repository, state, getUser(request.getEnvironment()), outputWriter);

            HookStdin stdin = request.getStdin();
            ReferenceTransaction transaction = stdin.readLines(parser);
            stdin.release();

            if (context != null) {
                if (shouldSuppressCallback(context, transaction, parser.isNonNullHashFound())) {
                    log.debug("[{}] Suppressing {} callback (last published state is {})",
                            repository, state, Objects.toString(context.publishedState, "-"));
                } else {
                    notifyListeners(context, transaction);
                }
                context.addCallback(transaction);
            } else {
                notifyListeners(null, transaction);
            }
            if (state == COMMITTED && !transaction.isRejected()) {
                UpdateRefsContext updateRefsContext = updateRefsContextHolder.get();
                if (updateRefsContext != null) {
                    updateRefsContext.markCommitted(transaction.getRefUpdates().stream()
                            .map(RefUpdate::getRefId)
                            .collect(toList()));
                }
            }
            return !(state == PREPARED && transaction.isRejected());
        } catch (IOException e) {
            // This should never happen when reading from a CharSequence
            log.warn("[{}] Error while parsing reference-transaction stdin", repository, e);
            return false;
        } finally {
            if (log.isTraceEnabled() && context != null) {
                String callbacks = context.callbackStates.stream()
                        .map(ReferenceTransactionState::toString)
                        .collect(Collectors.joining(", "));
                log.trace("[{}] callbacks: [{}], published: {}", repository, callbacks, context.publishedStates);
            }
            if (context != null && context.isFinished()) {
                transactionContext.remove();
            }
        }
    }

    @Override
    public void onHookContextClose() {
        ReferenceTransactionContext context = transactionContext.remove();
        if (context != null) {
            maybeClose(context);
        }
    }

    private static Person getUser(Map<String, String> environment) {
        String name = environment.get(GitConstants.ENV_AUTHOR_NAME);
        String email = environment.get(GitConstants.ENV_AUTHOR_EMAIL);
        if (name == null) {
            return null;
        }

        return new Person(name, email);
    }

    private static boolean hasReceivedAbortedOrPreparedCallback(ReferenceTransactionContext context) {
        return context.callbackStates.stream().anyMatch(state -> state == ABORTED || state == PREPARED);
    }

    private boolean isBrokenGitRefDelete(ReferenceTransactionContext context) {
        //in git versions lower than 2.37, git branch --delete and git tag --delete are subtly broken in how
        //the reference-transaction hook is invoked for the different refs backends. The normal (correct) flow is that
        //the PREPARED callbacks are done first for all affected backends, and only when they pass is the change
        //committed and are the COMMITTED callbacks invoked for all backends.
        //
        //For git branch --delete and git tag --delete, the behavior is broken; first git invokes PREPARED and COMMITTED
        //for the packed-refs backend, and only then does it call PREPARED, COMMITTED for each ref individually.
        //As a workaround, we need to publish the first (packed-refs) PREPARED callback because that is the only
        //one that is truly cancelable. And since that first PREPARED callback may contain multiple refs, we also
        //need to publish the corresponding (packed) COMMITTED callback, but we cannot publish it when we receive
        //the callback from git; at that time, there may still be loose refs that have not been deleted.
        //
        //Instead, we hold on to that COMMITTED callback and defer publishing it until we have received all COMMITTED
        //callbacks from the loose refs backend.
        return compatibilityConfig.isRefDeleteCallbacksBroken() &&
                ("branch".equals(context.command) || "tag".equals(context.command));
    }

    private void maybeClose(ReferenceTransactionContext inProgress) {
        if (inProgress.publishedState == PREPARED) {
            //Dangling transaction - listeners have been notified of PREPARED state, but not yet of the outcome
            ReferenceTransaction last = inProgress.last;
            if (inProgress.brokenGitRefDeleteWorkaround != null) {
                inProgress.brokenGitRefDeleteWorkaround.maybePublish();
                inProgress.brokenGitRefDeleteWorkaround = null;
            } else if (last.getState() == PREPARED) {
                log.warn("[{}] Ref update was PREPARED and {} but no {} callback was received",
                        inProgress.repository,
                        last.isRejected() ? "rejected" : "accepted",
                        last.isRejected() ? "ABORTED" : "COMMITTED");
            } else {
                log.info("[{}] Notifying listeners of {} outcome at the end of the execution context",
                        inProgress.repository, last.getState());
                notifyListeners(inProgress, last);
            }
        }
    }

    private void notifyListeners(ReferenceTransactionContext context, ReferenceTransaction transaction) {
        if (transaction.getRefUpdates().isEmpty()) {
            //Nothing to do
            return;
        }

        //Delegate transaction evaluation to the transaction listeners
        for (ReferenceTransactionListener listener : transactionListeners) {
            try {
                listener.onReferenceTransaction(transaction);
            } catch (RuntimeException e) {
                log.warn("[{}] {} failed while processing ref update ({})",
                        transaction.getRepository(), listener.getClass().getName(), transaction.getState(), e);
                if (transaction.getState() == PREPARED) {
                    transaction.reject(
                            "Internal error from reference-transaction hook. Check the Mesh logs for details");
                }
            }
        }
        if (context != null) {
            context.publishedState = transaction.getState();
            context.publishedStates.add(transaction.getState());
        }
    }

    /**
     * Detects 'shadow' reference-transaction callbacks that happen when one or more refs are deleted. When refs are
     * deleted, both packed-refs and loose refs may need to be updated. As a result, git triggers the
     * reference-transaction hook for both the packed and the loose ref backend, but the packed backend hook invocations
     * can be recognized because they only involve deleted refs AND the provided old-hash is always the NULL sha.
     * <p>
     * Consider a mixed update that creates branch A, updates branch U and deletes branch D, where U and D are packed.
     * In this scenario, git sends us the following callbacks:
     * <pre>{@code
     * 1: [prepared]
     * 0000000000000000000000000000000000000000 0000000000000000000000000000000000000000 D
     * 2: [prepared]
     * 0000000000000000000000000000000000000000 <new-a-hash> A
     * <old-d-hash> 0000000000000000000000000000000000000000 D
     * <old-u-hash> <new-u-hash> U
     * 3: [committed]
     * 0000000000000000000000000000000000000000 0000000000000000000000000000000000000000 D
     * 4: [committed]
     * 0000000000000000000000000000000000000000 <new-a-hash> A
     * <old-d-hash> 0000000000000000000000000000000000000000 D
     * <old-u-hash> <new-u-hash> U
     * }</pre>
     * Callbacks 1 and 3 correspond to updates to packed-refs. They only contain the ref deletions because
     * packed-refs does not need to be updated for updates.
     * <p>
     * If none of the deleted refs are packed in {@code packed-refs}, the callbacks look a bit different:
     * <pre>{@code
     * 1: [aborted]
     * 0000000000000000000000000000000000000000 0000000000000000000000000000000000000000 D
     * 2: [prepared]
     * 0000000000000000000000000000000000000000 <new-a-hash> A
     * <old-d-hash> 0000000000000000000000000000000000000000 D
     * <old-u-hash> <new-u-hash> U
     * 4: [committed]
     * 0000000000000000000000000000000000000000 <new-a-hash> A
     * <old-d-hash> 0000000000000000000000000000000000000000 D
     * <old-u-hash> <new-u-hash> U
     * }</pre>
     * Here, {@code git} has acquired a lock on {@code packed-refs}, then discovers that the {@code packed-refs} does
     * not need to be updated, releases the lock and invokes the ABORTED callback.
     */
    private boolean shouldSuppressCallback(ReferenceTransactionContext context,
                                           ReferenceTransaction transaction,
                                           boolean nonNullHashFound) {
        if (compatibilityConfig.isPackedRefsBackendCallbacksSuppressed()) {
            return false;
        }

        //When one or more refs are deleted, git locks the packed-refs to check whether the one of the refs that is
        //being deleted is packed. If that's the case, a PREPARED callback is sent for the packed-refs backend right
        //before the PREPARED callback is sent for the loose refs backend. The latter callback is complete; it contains
        //all refs, whether they are loose or not. So our goal here is to detect the packed-refs callback so it can be
        //ignored. Any 'fake' packed-refs callback will have only have deletes, with both the from-hash and to-hash
        //set to the NULL SHA. ReferenceTransactionParser will already have resolved these NULL from-hash to their
        //actual values, but we can use the nonNullHashFound flag as a short-cut.
        if (nonNullHashFound) {
            //At least one create/update/non-force delete detected. This cannot be a 'shadow' callback from packed-refs
            context.nonNullHashFound = true;
            return false;
        }

        Repository repository = transaction.getRepository();
        ReferenceTransactionState state = transaction.getState();
        if (context.brokenGitRefDeleteWorkaround != null) {
            return context.brokenGitRefDeleteWorkaround.onCallback(transaction);
        }

        //There are 4 scenarios for the FIRST callback:
        //1. an ABORTED callback from the packed-refs backend if packed-refs was locked, but none of the deleted refs
        //   were packed.
        //2. a PREPARED callback from the packed-refs backend with _only_ deletions.
        //3. a PREPARED callback from the packed-refs backend with _only_ creations/force updates. This is happens
        //   when pack-refs moves loose refs into packed-refs. For these updates, oldHash resolves to the same value
        //   as newHash and the ref-update is not added to transaction (because hash->hash is not an update). So, in
        //   this case, transaction.getRefUpdates() is an empty list.
        //4. a PREPARED callback from the loose refs backend without any deletions.
        if (transaction.getRefUpdates().size() < context.largestCallbackSize) {
            log.debug("[{}] partial {} callback detected from the packed-refs backend", repository, state);
            return true;
        }

        //At this point, we know that we received only DELETES. If there are any deletes, there will always be either
        //a PREPARED callback and/or an ABORTED callback for the packed-refs callback before we get callbacks for the
        //loose ref backend. Note that the loose ref transaction callbacks happen even if there is no loose ref.
        //Unfortunately, git invokes the callback in slightly different order depending on how refs are deleted:
        //
        // - update-ref (loose)             : aborted(fake), prepared(real), committed(real)
        // - update-ref (packed)            : prepared(f), prepared(r), committed(f), committed(r)
        // - update-ref (loose & packed)    : prepared(f), prepared(r), committed(f), committed(r)
        // - branch|tag -d (loose)          : prepared(r), committed(f), aborted(f), prepared(f), committed(r)
        // - branch|tag -d (packed)         : prepared(r), committed(f), aborted(f), prepared(f), committed(r)
        // - branch|tag -d (loose & packed) : prepared(r), committed(f), aborted(f), prepared(f), committed(r)
        // - push (loose)                   : aborted(f), prepared(r), committed(r)
        // - push (packed)                  : prepared(f), prepared(r), committed(f), committed(r)
        // - push (loose & packed)          : prepared(f), prepared(r), committed(f), committed(r)
        //
        //Note that for the packed-refs callbacks, the from-hashes are always NULL and it has been suggested on the
        //git mailing list that it's safe to ignore callbacks that have NULL from and to hashes for all ref updates.
        //Unfortunately, this is not true; forced updates also result in NULL from hashes. We could conclude the
        //reverse though: if we receive non-NULL from hashes, it must be the real thing. But because we're using
        //heuristics to detect which refs backend triggered the callback, we may get it wrong and end up invoking
        //identical callbacks multiple times. So we don't take this into account.
        //
        //To prevent duplicate or out-of-order callbacks we'll guard the invariant that
        //- we only publish a single prepared callback
        //- we only publish a single 'end state' callback (committed or aborted)
        //- we only publish an end state callback after publishing a prepared callback
        boolean preparedPublished = context.publishedStates.contains(PREPARED);
        if (state == PREPARED) {
            if (preparedPublished) {
                log.debug("[{}] duplicate PREPARED callback detected from the {} backend", repository,
                        context.brokenGitRefDeleteWorkaround == null ? "packed-refs" : "loose-refs");
                return true;
            }
            if (context.callbacksMatch(ABORTED) || context.callbacksMatch(PREPARED)) {
                //- aborted(f), __prepared(r)__, committed(r)
                //- prepared(f), __prepared(r)__, committed(f), committed(r)
                return false;
            }
            if (context.callbackStates.isEmpty() && isBrokenGitRefDelete(context)) {
                //- __prepared(r)__, committed(f), aborted(f), prepared(f), committed(r)
                context.brokenGitRefDeleteWorkaround = new BrokenGitRefDeleteWorkaround(context);
                return false;
            }
        } else if (!preparedPublished) {
            log.debug("[{}] {} callback detected from the packed-refs backend (prepared not yet processed)",
                    repository, state);
            return true;
        } else {
            //state is either COMMITTED or ABORTED and we've already published PREPARED. Suppress the callback
            //if we've already published an "end state"
            boolean abortedPublished = context.publishedStates.contains(ABORTED);
            boolean committedPublished = context.publishedStates.contains(COMMITTED);
            if (abortedPublished || committedPublished) {
                log.debug("[{}] {} callback detected from the packed-refs backend after {} callback",
                        repository, state, abortedPublished ? "ABORTED" : "COMMITTED");
                return true;
            }

            //This is an 'end state' callback; ABORTED or COMMITTED. In most cases, there will be two callbacks
            //and we should only respond to the second. But because there are subtle differences in the transaction
            //callback flows for different git commands, we need to be careful to not suppress the only 'real'
            //COMMITTED callback. So we do an extra check here to verify that all refs have actually been deleted and
            //therefore the changes have been finalized.
            if (state == COMMITTED) {
                if ((context.nonNullHashFound && !context.callbackStates.contains(COMMITTED)) ||
                        context.callbacksMatch(PREPARED, PREPARED)) {
                    //some previous callback(s) received non-null hashes, which means that we're not dealing with
                    //a force-update. Therefore, the current COMMITTED callback (with only 0000 hashes) must be fake.
                    log.debug("[{}] COMMITTED callback detected from the packed-refs backend", repository);
                    return true;
                }
            }
        }

        //Based on the reference-transaction invocations for the different git commands, we can conclude that we
        //always receive either an aborted or a prepared callback from the packed-refs backend before we receive a
        //'real' prepared callback
        if (hasReceivedAbortedOrPreparedCallback(context)) {
            return false;
        }
        log.debug("[{}] {} detected from the packed-refs backend", repository, state);
        return true;
    }

    private class BrokenGitRefDeleteWorkaround {

        private final ReferenceTransactionContext context;
        private volatile ReferenceTransaction deferredCommittedCallback;
        private volatile int looseRefCallbacksRemaining;

        BrokenGitRefDeleteWorkaround(ReferenceTransactionContext context) {
            this.context = context;
        }

        /**
         * @param transaction the reference-transaction callback
         * @return {@code true} if the reference-transaction should be suppressed
         */
        boolean onCallback(ReferenceTransaction transaction) {
            //Invariant: a BrokenGitRefDeleteWorkaround instance exists, so we must have published a PREPARED but not
            //yet published a COMMITTED or ABORTED callback.
            switch (transaction.getState()) {
                case ABORTED:
                    if (deferredCommittedCallback == null) {
                        //We haven't received a COMMITTED callback from the packed-refs backend yet so the initial
                        //published PREPARED must have been rejected and this ABORTED callback is real.
                        return false;
                    }
                    log.debug("[{}] ABORTED callback detected from the packed-refs backend (deferred = true)",
                            transaction.getRepository());
                    return true;
                case COMMITTED:
                    if (deferredCommittedCallback == null) {
                        //This is the COMMITTED packed-refs callback that we must capture and defer publishing until
                        //all COMMITTED callbacks from the loose-refs backend have been received.
                        deferredCommittedCallback = transaction;
                        looseRefCallbacksRemaining = transaction.getRefUpdates().size();
                        log.debug("[{}] Deferring COMMITTED callback from the packed-refs backend ({} deleted refs)",
                                transaction.getRepository(), looseRefCallbacksRemaining);
                        return true;
                    }
                    //This is a loose ref callback
                    if (--looseRefCallbacksRemaining == 0) {
                        //We've received loose-ref COMMITTED callbacks for all deleted refs
                        maybePublish();
                    }
                    log.debug("[{}] COMMITTED callback detected from the loose-refs backend (deferred = true)",
                            transaction.getRepository());
                    return true;
                default:
                    log.debug("[{}] PREPARED callback detected from the loose-refs backend (deferred = true)",
                            transaction.getRepository());
                    return true;
            }
        }

        void maybePublish() {
            if (deferredCommittedCallback != null) {
                log.debug("[{}] Publishing deferred COMMITTED callback", deferredCommittedCallback.getRepository());
                notifyListeners(context, deferredCommittedCallback);
            }
        }
    }

    /**
     * Represents the ref-update state constructed from multiple reference-transaction callbacks for the same
     * transaction. This context is used for two purposes:
     * <ul>
     *     <li>To detect cases where TWO (near) identical callbacks are received for the same update. This happens
     *      when a ref update affects both loose and packed refs (i.e. a ref deletion).</li>
     *      <li>To cache resolved refs in the case of forced ref updates and/or ref creation. In both these cases,
     *      the reference-transaction is called with oldHash set to 0000000000000000000000000000000000000000.
     *      We resolve the ref-id in question to determine the real oldHash in order to provide the
     *      ReferenceTransactionListeners with the correct value.</li>
     * </ul>
     */
    private static class ReferenceTransactionContext {

        private final List<ReferenceTransactionState> callbackStates;
        private final String command;
        private final Set<ReferenceTransactionState> publishedStates;
        private final Map<String, RefUpdate> refUpdates;
        private final Repository repository;

        private volatile BrokenGitRefDeleteWorkaround brokenGitRefDeleteWorkaround;
        private volatile int largestCallbackSize;
        private volatile ReferenceTransaction last;
        private volatile boolean nonNullHashFound;
        private volatile ReferenceTransactionState publishedState;

        ReferenceTransactionContext(Repository repository, String command) {
            this.command = command;
            this.repository = repository;

            callbackStates = new ArrayList<>();
            publishedStates = new LinkedHashSet<>(); //preserve the order
            refUpdates = new HashMap<>();
        }

        void addCallback(ReferenceTransaction callback) {
            last = callback;
            callbackStates.add(callback.getState());
            largestCallbackSize = Math.max(largestCallbackSize, callback.getRefUpdates().size());
        }

        boolean callbacksMatch(ReferenceTransactionState... states) {
            if (states.length != callbackStates.size()) {
                return false;
            }
            int i = 0;
            for (ReferenceTransactionState state : callbackStates) {
                if (state != states[i++]) {
                    return false;
                }
            }
            return true;
        }

        boolean isFinished() {
            return publishedStates.contains(ABORTED) || publishedStates.contains(COMMITTED);
        }
    }

    private class ReferenceTransactionParser extends AbstractRefUpdateParser<ReferenceTransaction> {

        private final ReferenceTransaction.Builder builder;
        private final ReferenceTransactionContext context;
        private final Map<String, RefUpdate.Builder> pending;
        private final Repository repository;

        private boolean nonNullHashFound;

        ReferenceTransactionParser(ReferenceTransactionContext context, Repository repository,
                                   ReferenceTransactionState state, Person user, GitHookOutputWriter writer) {
            this.context = context;
            this.repository = repository;

            builder = new ReferenceTransaction.Builder(repository, state, writer)
                    .user(user);
            pending = new HashMap<>();
        }

        @Nonnull
        @Override
        public ReferenceTransaction getResult() {
            if (!pending.isEmpty()) {
                Map<String, GitObjectInfo> resolved = builderFactory.builder(repository)
                        .catFile()
                        .batchCheck()
                        //Use a long timeout because we'll use the process to resolve refs on-demand.
                        //Note that the actual timeout will be capped by the RPC deadline, if present.
                        .withTimeout(Duration.ofHours(1))
                        .build(new InfoBatchCheckCatFilesStdioHandler(pending.keySet()))
                        .run();

                for (Map.Entry<String, RefUpdate.Builder> entry : pending.entrySet()) {
                    GitObjectInfo objectInfo = resolved.get(entry.getKey());
                    RefUpdate refUpdate = entry.getValue()
                            .oldHash(objectInfo == null ? ShaUtils.NULL_SHA1 : objectInfo.getObjectId())
                            .build();
                    if (context != null) {
                        context.refUpdates.put(entry.getKey(), refUpdate);
                    }
                    maybeAdd(refUpdate);
                }
                pending.clear();
            }

            return builder.build();
        }

        @Override
        protected boolean processRefUpdate(String refId, String oldHash, String newHash) {
            log.trace("[{}] {} {} {}", repository, oldHash, newHash, refId);
            if ("HEAD".equals(refId) || "FETCH_HEAD".equals(refId)) {
                //Ignore updates to symbolic refs like HEAD or FETCH_HEAD
                return true;
            }

            if (ShaUtils.isNull(oldHash)) {
                //a NULL oldHash means that the ref update is either a create operation, or a force update.
                //We need to resolve the ref to know for sure.
                RefUpdate existing = context.refUpdates.get(refId);
                if (existing == null) {
                    existing = resolveFromUpdateRefsContext(refId);
                }
                if (existing == null || isIncompatible(existing, oldHash, newHash)) {
                    //For efficiency, we postpone the resolving until the build() method is called
                    //so we can resolve them all with a single cat-file --batch-check operation.
                    pending.put(refId, new RefUpdate.Builder(refId).newHash(newHash));
                } else {
                    maybeAdd(existing);
                }
                nonNullHashFound = nonNullHashFound || !ShaUtils.isNull(newHash);
                return true;
            }

            nonNullHashFound = true;
            RefUpdate refUpdate = context.refUpdates.compute(refId, (id, existing) -> {
                if (existing == null || isIncompatible(existing, oldHash, newHash)) {
                    return new RefUpdate.Builder(refId)
                            .newHash(newHash)
                            .oldHash(oldHash)
                            .build();
                }
                return existing;
            });
            maybeAdd(refUpdate);

            return true;
        }

        boolean isNonNullHashFound() {
            return nonNullHashFound;
        }

        private boolean isIncompatible(RefUpdate existing, String oldHash, String newHash) {
            return !ShaUtils.hashesMatch(existing.getNewHash(), newHash) ||
                    (!ShaUtils.isNull(oldHash) && !ShaUtils.hashesMatch(existing.getOldHash(), oldHash));
        }

        private void maybeAdd(RefUpdate refUpdate) {
            //Only add the ref update if it describes a change. a x->x ref update can be triggered by
            //{@code git pack-refs} when loose refs are moved from into {@code packed-refs}.
            //Note that we always want to add
            String newHash = refUpdate.getNewHash();
            if (!ShaUtils.hashesMatch(refUpdate.getOldHash(), newHash)) {
                builder.refUpdates(refUpdate);
            }
        }

        private RefUpdate resolveFromUpdateRefsContext(String refId) {
            UpdateRefsContext context = updateRefsContextHolder.get();
            return context == null ? null : context.getRefUpdates().get(refId);
        }
    }
}
