package com.atlassian.bitbucket.mesh.git.hook.handler;

import com.atlassian.bitbucket.mesh.config.HooksConfig;
import com.atlassian.bitbucket.mesh.git.builder.GitProcess;
import com.atlassian.bitbucket.mesh.git.builder.GitProcessBuilder;
import com.atlassian.bitbucket.mesh.git.builder.GitProcessBuilderFactory;
import com.atlassian.bitbucket.mesh.git.builder.packobjects.PackObjectsStdioHandler;
import com.atlassian.bitbucket.mesh.git.cache.PackCacheService;
import com.atlassian.bitbucket.mesh.git.hook.GitHookHandler;
import com.atlassian.bitbucket.mesh.git.hook.GitHookOutputWriter;
import com.atlassian.bitbucket.mesh.git.hook.HookRequest;
import com.atlassian.bitbucket.mesh.git.hook.HookStdin;
import com.atlassian.bitbucket.mesh.repository.Repository;
import com.atlassian.bitbucket.mesh.throttle.ThrottleService;
import com.atlassian.bitbucket.mesh.throttle.ThrottledResource;
import com.atlassian.bitbucket.mesh.throttle.Ticket;
import com.atlassian.util.contentcache.CacheAccess;
import com.atlassian.util.contentcache.CacheResult;
import com.atlassian.util.contentcache.ContentCache;
import com.atlassian.util.contentcache.ContentProvider;
import com.atlassian.util.profiling.Ticker;
import com.atlassian.util.profiling.Timers;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
import com.google.common.hash.HashFunction;
import com.google.common.hash.Hasher;
import com.google.common.hash.Hashing;
import com.google.common.io.LineProcessor;
import org.apache.commons.codec.binary.Hex;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.io.IOException;
import java.io.OutputStream;
import java.io.UncheckedIOException;
import java.nio.ByteBuffer;
import java.util.Date;
import java.util.List;
import java.util.Map;

import static com.atlassian.bitbucket.mesh.git.GitConstants.ENV_THROTTLED_RESOURCE;
import static com.atlassian.bitbucket.mesh.git.hook.GitHook.PACK_OBJECTS;
import static com.atlassian.bitbucket.mesh.git.hosting.HostingOperationType.*;
import static java.nio.channels.Channels.newChannel;
import static java.util.Objects.requireNonNull;
import static org.apache.commons.lang3.StringUtils.strip;

/**
 * Caching {@link GitHookHandler} for {@code git-pack-objects} hook
 * <p>
 * Attempts to stream pack files from Content Cache, runs Git directly in case of {@code CACHE_MISS} or
 * {@code CACHE_BYPASS} response from Content Cache.
 * <p>
 * {@code git-pack-objects} delivers output on two streams - {@code stderr} and {@code stdout}. The output is
 * multiplexed into single cache entry using hook wire protocol.
 */
public class CachingPackObjectsHookHandler extends AbstractGitHookHandler {

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

    private final GitProcessBuilderFactory builderFactory;
    private final HooksConfig config;
    private final PackCacheService packCacheService;
    private final ThrottleService throttleService;

    public CachingPackObjectsHookHandler(GitProcessBuilderFactory builderFactory, HooksConfig config,
                                         PackCacheService packCacheService, ThrottleService throttleService) {
        super(PACK_OBJECTS);

        this.builderFactory = builderFactory;
        this.config = config;
        this.packCacheService = packCacheService;
        this.throttleService = throttleService;
    }

    @Override
    public boolean handle(@Nonnull Repository repository, @Nonnull HookRequest request,
                          @Nonnull GitHookOutputWriter outputWriter) {
        requireNonNull(repository, "repository");
        requireNonNull(request, "request");
        requireNonNull(outputWriter, "outputWriter");

        return maybeHandle(new Context(request), repository, outputWriter);
    }

    @VisibleForTesting
    String calculateKey(List<String> arguments, HookStdin stdin) {
        HashFunction sha256 = Hashing.sha256();

        Hasher hasher = sha256.newHasher();
        arguments.forEach(argument -> {
            hasher.putUnencodedChars(argument);
            hasher.putByte((byte) 0x00); //Separate arguments with a NUL byte
        });

        //A hash of the command line arguments serves as the base for the cache key. We'll merge hashes
        //for the provided "want" SHAs into this to compute the final result
        byte[] result = hasher.hash().asBytes();

        //Just because ref advertisements serve the same refs with the same hashes does _not_ mean wants
        //are provided to the pack-objects callback in the same order. This is particularly likely when
        //using different protocols (e.g. HTTP and SSH). Each may provide the refs in a consistent order
        //but they may not be the _same_ order, resulting in unnecessary cache misses. This approach of
        //combining hashes, rather than generating a single hash over everything, is intended to produce
        //the same hash, given the same wants, regardless of their order (See Hashing.combineUnordered).
        //We do it this way, rather than sorting the wants, to avoid having them all in memory
        try {
            return stdin.readLines(new LineProcessor<String>() {

                @Override
                public String getResult() {
                    return Hex.encodeHexString(result);
                }

                @Override
                public boolean processLine(@Nonnull String line) {
                    byte[] hash = sha256.newHasher().putUnencodedChars(line).hash().asBytes();
                    for (int i = 0; i < hash.length; ++i) {
                        result[i] += hash[i];
                    }

                    return true;
                }
            });
        } catch (IOException e) {
            throw new UncheckedIOException("Failed to generate cache key; stdin could not be read", e);
        }
    }

    @VisibleForTesting
    PackObjectsStdioHandler newHandler(HookStdin stdin, OutputStream cache) {
        GitHookOutputWriter cacheWriter = new GitHookOutputWriter(newChannel(cache)) {

            @Override
            public void writeStderr(@Nonnull ByteBuffer buffer) throws IOException {
                //stderr contains progress information that accumulates slowly, but lets clients know the
                //request is progressing. To avoid long delays on the client before seeing progress from
                //the server, eagerly flush every stderr buffer to the cache stream
                super.writeStderr(buffer);
                cache.flush();
            }
        };

        return new PackObjectsStdioHandler(stdin, cacheWriter);
    }

    @VisibleForTesting
    boolean shouldCache(HookStdin stdin) {
        try {
            return stdin.readLines(new LineProcessor<Boolean>() {

                private boolean foundNot;
                private Boolean result = Boolean.TRUE; //Assume we should cache until we find a have

                @Override
                public Boolean getResult() {
                    return result;
                }

                @Override
                public boolean processLine(@Nonnull String line) {
                    if (foundNot && !line.isEmpty()) {
                        //SHAs after --not represent haves, so this is a fetch rather than a clone
                        result = Boolean.FALSE;
                        return false;
                    }
                    if ("--not".equals(line)) {
                        foundNot = true;
                    }

                    return true;
                }
            });
        } catch (IOException e) {
            throw new UncheckedIOException("Failed to parse stdin to decide whether to cache or not", e);
        }
    }

    @Nullable
    private String getCacheKey(Context context, Repository repository) {
        try {
            return calculateKey(context.getArguments(), context.getStdin());
        } catch (Exception e) {
            log.warn("{}: Unable to calculate cache key", repository, e);
            return null;
        }
    }

    private boolean maybeHandle(Context context, Repository repository, GitHookOutputWriter outputWriter) {
        //Release the current ticket if one is held
        throttleService.getTicket().ifPresent(Ticket::close);

        if (context.isFiltered() || !shouldCache(context.getStdin())) {
            outputWriter.setFilter(context.getFilter()); //Filtering can be applied regardless of shallowness
            outputWriter.setOperationType(context.isShallow() ? SHALLOW_FETCH : FETCH);
            if (log.isTraceEnabled()) {
                log.trace("{}: '{}' looks like a fetch; will not cache. Stdin:\n{}",
                        repository, context.getArguments(), context.getStdin());
            }

            //Fall through to streamFromGit
        } else {
            outputWriter.setOperationType(context.isShallow() ? SHALLOW_CLONE : CLONE);
            if (log.isTraceEnabled()) {
                log.trace("{}: '{}' looks like a clone; will try using cache. Stdin:\n{}",
                        repository, context.getArguments(), context.getStdin());
            }

            String key = getCacheKey(context, repository);
            if (key != null) {
                return streamFromCache(context, repository, key, outputWriter);
            }
        }

        return streamFromGit(context, repository, new PackObjectsStdioHandler(context.getStdin(), outputWriter));
    }

    private ContentProvider newContentProvider(Context context, Repository repository) {
        return new ContentProvider() {

            @Override
            public void apply(OutputStream cache) throws IOException {
                streamFromGit(context, repository, newHandler(context.getStdin(), cache));
                cache.flush();
            }

            @Override
            public Date getExpiry() {
                return packCacheService.createExpiryDate();
            }
        };
    }

    private boolean streamFromCache(Context context, Repository repository, String cacheKey,
                                    GitHookOutputWriter outputWriter) {
        //This must use ContentCache.access, not accessStream! access uses a background thread to pump data
        //from cache entries that are still being populated, to avoid having clients sit idle until the full
        //pack is spooled to the cache. accessStream blocks until the cache entry is fully populated.
        ContentCache cache = packCacheService.getCache(repository);
        try (CacheAccess access = cache.access(cacheKey, outputWriter.asOutputStream(),
                newContentProvider(context, repository))) {
            CacheResult result = access.getResult();
            outputWriter.setCacheResult(result);

            if (result == CacheResult.BYPASS) {
                log.warn("{}: Bypassing the cache", repository);
                //Fall-through to stream from Git
            } else {
                log.debug("{}: Attempt to stream from cache resulted in [{}]", repository, result);
                if (result == CacheResult.HIT) {
                    //If we're getting a cache hit, we know we don't need the stdin that was provided to the
                    //hook--we're going to stream an existing pack from disk
                    context.getStdin().release();
                }

                try (Ticker ignored = Timers.start(result + " [" + cacheKey + "]")) {
                    access.stream();
                    return true;
                } catch (IOException e) {
                    throw new UncheckedIOException("Streaming pack-objects from cache failed", e);
                }
            }
        } catch (IOException e) {
            log.error("{}: Failed to set up cache access", repository, e);
            //Fall-through to stream from Git
        }

        return streamFromGit(context, repository, new PackObjectsStdioHandler(context.getStdin(), outputWriter));
    }

    private boolean streamFromGit(Context context, Repository repository,
                                  PackObjectsStdioHandler gitHandler) {
        GitProcessBuilder builder = builderFactory.builder(repository);

        GitProcess<Void> process = context.prepareBuilder(builder).build(gitHandler);
        process.setTimeout(config.getPackObjectsTimeout());
        process.run();

        return true;
    }

    private static class Context {

        private final List<String> arguments;
        private final Map<String, String> environment;
        private final HookStdin stdin;

        private String filter;
        private boolean shallow;

        Context(@Nonnull HookRequest request) {
            ImmutableList.Builder<String> builder = ImmutableList.builder();
            request.getArguments().forEach(argument -> {
                if ("--shallow".equals(argument)) {
                    shallow = true;
                } else if (argument.startsWith("--filter=")) {
                    //Git calls the pack-objects hook with arguments like --filter='blob:none'. We're
                    //not a shell, though, so we don't need shell escapes and leaving them in causes
                    //the pack-objects process to fail
                    filter = strip(argument.substring(9), "'");
                    argument = "--filter=" + filter; //Replace the argument with our cleaned version
                }

                builder.add(argument);
            });

            arguments = builder.build();
            environment = request.getEnvironment();
            stdin = request.getStdin();
        }

        @Nonnull
        List<String> getArguments() {
            return arguments;
        }

        @Nullable
        String getFilter() {
            return filter;
        }

        @Nonnull
        HookStdin getStdin() {
            return stdin;
        }

        boolean isFiltered() {
            return filter != null;
        }

        boolean isShallow() {
            return shallow;
        }

        @Nonnull
        GitProcessBuilder prepareBuilder(@Nonnull GitProcessBuilder builder) {
            //Set the correct resource for throttling. If one isn't specified, default to PACK
            String resource = environment.getOrDefault(ENV_THROTTLED_RESOURCE, ThrottledResource.PACK.getId());
            builder.throttled(ThrottledResource.forId(resource));

            arguments.stream()
                    .skip(1) //Drop the leading "git", which will always be the first argument
                    .forEach(builder::rawArgument); //Use rawArgument to apply Git's arguments verbatim
            environment.forEach(builder::withEnvironment); //Copy Git's environment variables

            return builder;
        }
    }
}
