package com.atlassian.bitbucket.mesh.git.builder.fetch;

import com.atlassian.bitbucket.mesh.git.GitAgent;
import com.atlassian.bitbucket.mesh.git.exception.CommandFailedException;
import com.atlassian.bitbucket.mesh.git.exception.RemoteAuthenticationException;
import com.atlassian.bitbucket.mesh.git.exception.RemoteAuthorizationException;
import com.atlassian.bitbucket.mesh.git.fetch.FetchCallback;
import com.atlassian.bitbucket.mesh.git.ref.Branch;
import com.atlassian.bitbucket.mesh.git.ref.Ref;
import com.atlassian.bitbucket.mesh.git.ref.RefUpdate;
import com.atlassian.bitbucket.mesh.process.OsProcess;
import com.atlassian.bitbucket.mesh.process.ProcessCompletionHandler;
import com.atlassian.bitbucket.mesh.process.ProcessSummary;
import com.atlassian.bitbucket.mesh.process.handler.AbstractProcessStdioHandler;
import com.atlassian.bitbucket.mesh.process.handler.LineDecoder;
import com.atlassian.bitbucket.mesh.process.handler.LineOutputAdapter;
import com.atlassian.bitbucket.mesh.repository.Repository;
import com.google.common.annotations.VisibleForTesting;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.*;

import static com.atlassian.bitbucket.mesh.git.GitRefPattern.HEADS;
import static com.atlassian.bitbucket.mesh.git.GitRefPattern.TAGS;
import static com.atlassian.bitbucket.mesh.util.ShaUtils.NULL_SHA1;

/**
 * Handler for a {@code git fetch} command which parse out the added, updated and deleted refs and pass them to the
 * provided {@link FetchCallback}.
 * <p>
 * The parsing applied here makes a few assumptions on how the command is executed:
 * <ul>
 *     <li>{@code --prune} may be used, so branches and tags may be added or deleted</li>
 *     <li>
 *         The refspec applied will copy refs <i>to the same place</i>, meaning refs/heads/master from upstream
 *         will update refs/heads/master in this repository
 *     </li>
 * </ul>
 * Note that {@link FetchStdioHandler} does <em>not</em> call {@link FetchCallback#onStart()} or
 * {@link FetchCallback#onEnd(ProcessSummary)} because a fetch operation may be split into a number of fetches to
 * avoid exceeding command line limits. When this happens, the caller should ensure that {@code onStart} and
 * {@code onEnd} are called exactly once.
 */
public class FetchStdioHandler extends AbstractProcessStdioHandler<ProcessSummary> implements ProcessCompletionHandler {

    private static final int MAX_ERROR_TEXT_SIZE = 64 * 1024;
    private static final Logger log = LoggerFactory.getLogger(FetchStdioHandler.class);

    private final List<String> addedRefs;
    private final GitAgent agent;
    private final FetchCallback callback;
    private final ErrorText errors;
    private final Map<String, Ref> preFetchRefs;
    private final String remoteUrl;
    private final Repository repository;
    private final LineOutputAdapter stderrAdapter;
    private final LineOutputAdapter stdoutAdapter;

    private boolean anyRefUpdated;
    private int refCount;
    private ProcessSummary summary;

    public FetchStdioHandler(GitAgent agent, FetchCallback callback, List<Ref> refs, String remoteUrl,
                             Repository repository) {
        this.agent = agent;
        this.callback = callback;
        this.remoteUrl = remoteUrl;
        this.repository = repository;

        addedRefs = new ArrayList<>();
        errors = new ErrorText();
        preFetchRefs = new HashMap<>();
        refs.forEach(ref -> {
            preFetchRefs.putIfAbsent(unqualify(ref.getId()), ref);
            preFetchRefs.put(ref.getId(), ref);
        });
        //Fetch sends progress information and errors to stderr, so we redirect both stdout and stderr processing to
        //onLine, but use separate LineOutputAdapter instances for them to make sure we don't mix bytes from different
        //streams in a single line. We can't simply use the same adapter instance because the adapter buffers internally
        //to handle split multibyte characters and split lines. Mixing bytes from different unrelated streams can result
        //in garbage output.
        LineOutputAdapter.Builder adapterBuilder = LineOutputAdapter.newBuilder()
                .charset(StandardCharsets.UTF_8)
                .lineMode(LineDecoder.Mode.CRLF_OR_LF)
                .lineSink(this::onLine);
        stderrAdapter = adapterBuilder.build();
        stdoutAdapter = adapterBuilder.build();
    }

    @Override
    public ProcessSummary getResult() {
        return summary;
    }

    public boolean isAnyRefUpdated() {
        return anyRefUpdated;
    }

    @Override
    public void onComplete(@Nonnull ProcessSummary summary) {
        if (summary.getExitCode() != 0) {
            errors.getAuthenticationError().ifPresent(err -> {
                throw new RemoteAuthenticationException(err, remoteUrl);
            });
            errors.getAuthorisationOrVisibilityError().ifPresent(err -> {
                throw new RemoteAuthorizationException(err, remoteUrl);
            });
            this.summary = new ProcessSummary.Builder(summary)
                    .stderr(errors.toString())
                    .build();
            if (refCount == 0) {
                throw new CommandFailedException("Fetch failed", this.summary);
            }
        } else {
            this.summary = summary;
        }
        wrapUp();
    }

    @Override
    public void onPreStart(@Nonnull OsProcess process) {
        stderrAdapter.attach(process);
        stdoutAdapter.attach(process);

        super.onPreStart(process);
    }

    @Override
    public void onStderr(@Nonnull ByteBuffer buffer, boolean closed) {
        stderrAdapter.onBytes(buffer, closed);
    }

    @Override
    public void onStdout(@Nonnull ByteBuffer buffer, boolean closed) {
        stdoutAdapter.onBytes(buffer, closed);
    }

    @VisibleForTesting
    String getErrors() {
        return errors.toString();
    }

    private static boolean isRefUpdateLine(String line) {
        //Blank lines aren't really expected, but are ignored if present. Sometimes "git fetch" outputs lines like:
        //  from ssh://stash.dev.internal.atlassian.com:7999/STASH/stash
        //While that is also not really expected here, such lines are ignored as well.
        return !StringUtils.isBlank(line) && line.length() >= 3 && line.charAt(0) == ' ' && line.charAt(2) == ' ';
    }

    private static String parseRefName(String line, String change) {
        String name = line.substring(change.length() + 3); // strip off the hint and the change
        int split = name.indexOf(" -> ");
        if (split != -1) {
            //Only look at the destination ref; it will _always_ match the source, but source is (none) for deleted refs
            name = name.substring(split + 4).trim();
        }

        // the change may have a hint often in parentheses e.g.
        // "6a1b128..2705cad  short-branch-name -> short-branch-name  (forced update)"
        // strip it out including the spaces
        int hintIndex = name.indexOf(' ');
        if (hintIndex != -1) {
            name = name.substring(0, hintIndex);
        }
        return name;
    }

    private static String parseRefType(String change, char hint) {
        if (change.contains("branch")) {
            return "branch";
        }
        if (change.contains("tag") || hint == 't') {
            return "tag";
        }
        if (hint == ' ') {
            // must be a branch update. Tag updates are listed differently
            return "branch";
        }
        return "ref";
    }

    private static String unqualify(String refId) {
        if (refId.startsWith(HEADS.getPath())) {
            return HEADS.unqualify(refId);
        }
        if (refId.startsWith(TAGS.getPath())) {
            return TAGS.unqualify(refId);
        }
        if (refId.startsWith("refs/") && refId.length() > 5) {
            return refId.substring(5);
        }
        return refId;
    }

    private boolean isProgressLine(String line) {
        if (refCount > 0) {
            return false;
        }
        return line.startsWith("remote: Counting objects") ||
                line.startsWith("remote: Compressing objects") ||
                line.startsWith("remote: Enumerating objects") ||
                line.startsWith("remote: Total") ||
                line.startsWith("Delta compression") ||
                line.startsWith("Receiving objects") ||
                line.startsWith("Resolving deltas") ||
                line.startsWith("Writing objects"); // we don't do these... yet;
    }

    private boolean onLine(@Nonnull String line, boolean truncated) {
        if (!isRefUpdateLine(line)) {
            if (!isProgressLine(line)) {
                //Since we are an error handler as well as an output handler, and git fetch outputs progress to stderr,
                //buffer up any lines that we don't process so that they can be provided to our exit handler side
                errors.add(line);
            }

            return true;
        }

        //Parse out the hint, which is always the second character in the line and one character long
        char hint = line.charAt(1); //Possible hints are *, x, -, ! and space

        //Parse out the change details
        int changeEnd = line.indexOf(' ', 20); //3 + 17, where 17 -> (7*2 + 3), where 7 is the short hash length
        String change = line.substring(3, changeEnd).trim(); //Some messages are shorter than the 17 minimum

        //Parse out the ref name
        String refName = parseRefName(line, change);
        String refType = parseRefType(change, hint);
        String refId = qualify(refName, refType);

        switch (hint) {
            case '*': //* identifies added refs
                // * [new branch]      short-branch-name -> short-branch-name
                // * [new tag]         short-tag-name -> short-tag-name
                // * [new ref]         short-ref-name -> short-ref-name
                //This ref did not exist within this repository and has been added.

                //git fetch emits its ref updates on stdout before it actually applies them to the repository, so
                //we cannot immediately resolve the new hash of added refs. Instead we remember which refs have been
                //added, resolve them in wrapUp after the process has completed and then notify the callback.
                addedRefs.add(refId);
                break;
            case '!': //This ref could not be updated.
                onRefFailure(refId);
                log.warn("{}: Ref {} could not be updated", repository, refName);
                break;
            case '-':
                // - [deleted]         (none)     -> short-branch-name (2.10+)
                onRefUpdate(refId, preFetchHash(refId), NULL_SHA1);
                break;
            case 't':
                // t [tag update]      retagged_signed_tag -> retagged_signed_tag (2.10+)
                onRefUpdate(refId, preFetchHash(refId), resolveRef(refId));
                break;
            case '=':
                //Ref is up to date, ignore
                log.debug("{}: ref {} already up to date", repository, refId);
                break;
            default:
                //   6a1b128..2705cad  short-branch-name -> short-branch-name
                //   OR
                // + 6a1b128..2705cad  short-branch-name -> short-branch-name (forced update)
                //
                //This branch already exists in the repository, so the branch has been updated to match the upstream
                int split = change.indexOf("..");
                String fromHash = change.substring(0, split);

                //There may be .. or ... between the two hashes
                split += 2;
                if (change.charAt(split) == '.') {
                    ++split;
                }
                String toHash = change.substring(split); //Drop the .. between the two hashes

                //This will build a RefUpdate that has short hashes for getFromHash() and getToHash(). That violates the
                //contract for RefUpdate, and is known to cause problems in some clients (Technical Insights for Stash,
                //for example). The RefUpdate constructed here will be replaced later with one using full hashes. It is
                //delayed out of here for two reasons:
                //1. To not delay processing for the git fetch process this handler is attached to
                //2. To allow _all_ such short hashes to be resolved in a single pass
                onRefUpdate(refId, fromHash, toHash);
        }
        return true;
    }

    private void onRefFailure(String refId) {
        refCount++;
        callback.onFailedUpdate(refId);
    }

    private void onRefUpdate(String refId, String oldHash, String newHash) {
        anyRefUpdated = true;
        refCount++;
        callback.onRefUpdate(RefUpdate.of(refId, oldHash, newHash));
    }

    private String preFetchHash(String refId) {
        Ref ref = preFetchRefs.get(refId);
        return ref == null ? null : ref.getLatestCommit();
    }

    private String qualify(String refName, String refType) {
        if ("FETCH_HEAD".equals(refName)) {
            return refName;
        }
        if (refType == null || "ref".equals(refType)) {
            if (refName.startsWith("refs/")) {
                return refName;
            }
            Ref existing = preFetchRefs.get(refName);
            if (existing != null) {
                return existing.getId();
            }
            return refName;
        }
        switch (refType) {
            case "branch":
                return HEADS.qualify(refName);
            case "tag":
                return TAGS.qualify(refName);
            default:
                return refName;
        }
    }

    private String resolveRef(String refId) {
        Branch ref = agent.resolveBranch(repository, refId, true);
        return ref != null ? ref.getLatestCommit() : null;
    }

    private void wrapUp() {
        for (String refId : addedRefs) {
            String newHash = resolveRef(refId);
            if (newHash == null) {
                log.warn("[{}] Failed to resolve added ref '{}'.", repository, refId);
            } else {
                onRefUpdate(refId, NULL_SHA1, newHash);
            }
        }
    }

    /**
     * A collection of error lines with a limit applied to the overall length to avoid leaking memory
     * through accidental accumulation of progress lines etc
     */
    private static class ErrorText {

        private final Deque<String> lines;

        private int length;

        ErrorText() {
            lines = new ArrayDeque<>();
        }

        public void add(String line) {
            if (line.isEmpty()) {
                return;
            }

            int newLineLength = line.length();
            while (!lines.isEmpty() && (length + newLineLength > MAX_ERROR_TEXT_SIZE)) {
                length -= lines.removeFirst().length();
            }
            lines.addLast(line);
            length += newLineLength;
        }

        @Override
        public String toString() {
            return String.join("\n", lines);
        }

        private Optional<String> getAuthenticationError() {
            return lines.stream()
                    .filter(it -> it.contains("Host key verification failed") || // SSH
                            it.contains("Permission denied (") ||          // SSH
                            it.contains("fatal: Authentication failed") || // HTTP
                            it.contains("fatal: could not read Password")) // HTTP - no password specified
                    .findFirst();
        }

        private Optional<String> getAuthorisationOrVisibilityError() {
            return lines.stream()
                    .filter(it -> it.contains("Could not read from remote repository") ||
                            it.contains("Repository not found"))
                    .findFirst();
        }
    }
}
