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

import com.atlassian.bitbucket.mesh.AbstractGrpcTest;
import com.atlassian.bitbucket.mesh.git.hosting.GitPacketStream;
import com.atlassian.bitbucket.mesh.git.hosting.GitPackets;
import com.atlassian.bitbucket.mesh.io.IoFunction;
import com.atlassian.bitbucket.mesh.junit.TemporaryDirectory;
import com.atlassian.bitbucket.mesh.repository.Repository;
import com.atlassian.bitbucket.mesh.rpc.v1.*;
import com.atlassian.bitbucket.mesh.test.annotation.Repo;
import com.atlassian.bitbucket.mesh.util.PackfileListener;
import com.google.common.collect.ImmutableList;
import com.google.protobuf.ByteString;
import io.grpc.ManagedChannel;
import io.grpc.stub.ClientCallStreamObserver;
import io.grpc.stub.ClientResponseObserver;
import org.apache.commons.compress.archivers.ArchiveEntry;
import org.apache.commons.compress.archivers.ArchiveInputStream;
import org.apache.commons.compress.archivers.tar.TarArchiveInputStream;
import org.apache.commons.compress.archivers.zip.ZipArchiveInputStream;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;

import javax.annotation.Resource;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.UncheckedIOException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

import static org.hamcrest.Matchers.equalTo;

/**
 * Verifies {@link RpcUploadArchiveRequest} without using {@code git archive} as a client.
 * <p>
 * {@code git upload-archive} reads and writes data in Git's {@code pkt-line} format (for format details,
 * <a href="https://github.com/git/git/blob/v2.27.0/Documentation/technical/protocol-common.txt">read this</a>).
 * That makes the tests here a little more complicated, because they need to parse the {@code pkt-line} format
 * so they can write the archive contents without unexpected {@code pkt-line} cruft.
 */
public class UploadArchiveRequestTest extends AbstractGrpcTest {

    @Rule
    public final TemporaryDirectory tempDir = new TemporaryDirectory();

    @Resource
    private ManagedChannel managedChannel;
    @Repo
    private Repository repository;
    private HostingServiceGrpc.HostingServiceStub stub;

    @Before
    public void setup() {
        stub = HostingServiceGrpc.newStub(managedChannel);
    }

    @Test
    public void testUploadArchive() throws Exception {
        Path archive = tempDir.newFile("repo.tar");
        String arguments = "0014argument master\n" +
                "0010argument --\n" +
                "0000";
        RpcUploadArchiveRequest request = RpcUploadArchiveRequest.newBuilder()
                .setRepository(repository.getId())
                .build();

        UploadArchiveResponseObserver responseObserver = new UploadArchiveResponseObserver(request, arguments, archive);

        stub.uploadArchive(responseObserver);
        //The default format is an uncompressed tarball
        RpcHostingStreamResult result = responseObserver.asResult();

        assertEquals(0, result.getExitCode());

        List<String> entries = listArchive(archive, TarArchiveInputStream::new);
        assertEquals(ImmutableList.of("add_file/add_file.txt",
                "basic_branching/file.txt",
                "binary/B.zip",
                "binary/C.zip",
                "binary/D.zip",
                "branch_mod_merge/file.txt",
                "modification/mod_file.txt",
                "moved_dir/file.txt",
                "mv_file/moved_file.txt",
                "out_of_order/file.txt",
                "symlink/link.txt",
                "symlink/target.txt"), entries);
        assertNotNull(result);
        assertThat(result.getOperationType(), equalTo(RpcHostingOperationType.HOSTING_OP_ARCHIVE));
        assertFalse(result.hasProtocolVersion());
    }

    @Test
    public void testUploadArchiveForZip() throws Exception {
        Path archive = tempDir.newFile("repo.zip");
        String arguments = "001aargument --format=zip\n" +
                "0014argument master\n" +
                "0010argument --\n" +
                "0014argument binary\n" +
                "0000";
        RpcUploadArchiveRequest request = RpcUploadArchiveRequest.newBuilder()
                .setRepository(repository.getId())
                .build();

        //Explicitly request a zipfile, and filter to only the files in the binary directory
        UploadArchiveResponseObserver responseObserver = new UploadArchiveResponseObserver(request, arguments, archive);

        stub.uploadArchive(responseObserver);
        RpcHostingStreamResult result = responseObserver.asResult();

        assertEquals(0, result.getExitCode());

        List<String> entries = listArchive(archive, ZipArchiveInputStream::new);
        assertEquals(ImmutableList.of("binary/B.zip", "binary/C.zip", "binary/D.zip"), entries);
        assertNotNull(result);
        assertThat(result.getOperationType(), equalTo(RpcHostingOperationType.HOSTING_OP_ARCHIVE));
        assertFalse(result.hasProtocolVersion());
    }

    private static List<String> listArchive(Path archive, IoFunction<InputStream, ArchiveInputStream> streamFactory)
            throws IOException {
        List<String> entries = new ArrayList<>();
        try (InputStream inputStream = Files.newInputStream(archive);
             ArchiveInputStream tar = streamFactory.apply(inputStream)) {
            ArchiveEntry entry;
            while ((entry = tar.getNextEntry()) != null) {
                if (entry.isDirectory() || "pax_global_header".equals(entry.getName())) {
                    continue;
                }
                entries.add(entry.getName());
            }
        }

        return entries;
    }

    private static class UploadArchiveResponseObserver
            implements ClientResponseObserver<RpcUploadArchiveFragment, RpcHostingReadStreamFragment> {

        private final Path archive;
        private final String arguments;
        private final RpcUploadArchiveRequest request;
        private final CompletableFuture<RpcHostingStreamResult> resultFuture;

        private GitPacketStream packetStream;
        private volatile ClientCallStreamObserver<RpcUploadArchiveFragment> requestStream;
        private RpcHostingStreamResult result;
        private OutputStream writer;

        UploadArchiveResponseObserver(RpcUploadArchiveRequest request, String arguments, Path archive) {
            this.archive = archive;
            this.arguments = arguments;
            this.request = request;

            resultFuture = new CompletableFuture<>();
        }

        @Override
        public void beforeStart(ClientCallStreamObserver<RpcUploadArchiveFragment> requestStream) {
            this.requestStream = requestStream;
        }

        @Override
        public void onCompleted() {
            resultFuture.complete(result);
        }

        @Override
        public void onError(Throwable t) {
            resultFuture.completeExceptionally(t);
        }

        @Override
        public void onNext(RpcHostingReadStreamFragment value) {
            switch (value.getFragmentOneofCase()) {
                case RESULT:
                    onResult(value.getResult());
                    break;
                case START:
                case STDERR:
                    break;
                case STDOUT:
                    onStdout(value.getStdout());
                    break;
            }
        }

        RpcHostingStreamResult asResult() throws ExecutionException, InterruptedException {
            //send the request
            requestStream.onNext(RpcUploadArchiveFragment.newBuilder()
                    .setRequest(request)
                    .build());
            //send stdin
            requestStream.onNext(RpcUploadArchiveFragment.newBuilder()
                    .setStdin(RpcStreamChunk.newBuilder()
                            .setClosed(true)
                            .setData(ByteString.copyFrom(arguments, StandardCharsets.UTF_8)))
                    .build());
            //wait for the result
            return resultFuture.get();
        }

        private void onResult(RpcHostingStreamResult result) {
            this.result = result;
        }

        private void onStdout(RpcStreamChunk chunk) {
            try {
                if (writer == null) {
                    writer = Files.newOutputStream(archive);

                    //Rather than duplicate logic for parsing Git's pkt-line format, we use the
                    //GitPacketStream to do the parsing for us to assemble full packets
                    packetStream = new GitPacketStream();
                    packetStream.addListener(
                            new PackfileListener(writer, ignored -> { }, (buf, closed) -> { }, GitPackets::isFlush));
                }
                for (ByteBuffer buffer : chunk.getData().asReadOnlyByteBufferList()) {
                    packetStream.append(buffer);
                }

                if (chunk.getClosed()) {
                    writer.close();
                }
            } catch (IOException e) {
                throw new UncheckedIOException(e);
            }
        }
    }
}
