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

import com.atlassian.bitbucket.mesh.AbstractGrpcTest;
import com.atlassian.bitbucket.mesh.git.builder.GitProcessBuilderFactory;
import com.atlassian.bitbucket.mesh.git.hosting.GitPacketStream;
import com.atlassian.bitbucket.mesh.git.hosting.GitPackets;
import com.atlassian.bitbucket.mesh.junit.TemporaryDirectory;
import com.atlassian.bitbucket.mesh.process.handler.StringStdioHandler;
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 io.grpc.ManagedChannel;
import io.grpc.stub.ClientCallStreamObserver;
import io.grpc.stub.ClientResponseObserver;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;

import javax.annotation.Resource;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
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 com.atlassian.bitbucket.mesh.rpc.util.ByteStringUtils.nullToEmpty;
import static org.hamcrest.Matchers.contains;
import static org.hamcrest.Matchers.containsString;

/**
 * Verifies {@link RpcUploadPackRequest} without using {@code git fetch-pack} as a client.
 * <p>
 * {@code git upload-pack} 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 process the pack without unexpected {@code pkt-line} cruft.
 */
public class UploadPackRequestTest extends AbstractGrpcTest {

    private static final String FLUSH_PACKET = new String(GitPackets.getFlush(), StandardCharsets.UTF_8);

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

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

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

    @Test
    public void testUploadPack() throws Exception {
        String arguments = "0090want 0a943a29376f2336b78312d99e65da17048951db multi_ack_detailed side-band-64k thin-pack ofs-delta deepen-since deepen-not agent=git/2.28.0\n" +
                "00000009done\n";
        Path pack = tempDir.newFile("test.pack");
        RpcUploadPackRequest request = RpcUploadPackRequest.newBuilder()
                .setRepository(repository.getId())
                .build();

        UploadPackResponseObserver responseObserver = new UploadPackResponseObserver(request, arguments, pack);
        stub.uploadPack(responseObserver);

        RpcHostingStreamResult result = responseObserver.asResult();
        assertNotNull(result);
        assertEquals(0, result.getExitCode());
        assertEquals(0, result.getProtocolVersion());
        assertSame(RpcPackCacheResult.PACK_CACHE_MISS, result.getCacheResult());
        assertSame(RpcHostingOperationType.HOSTING_OP_CLONE, result.getOperationType());
        assertRefs(responseObserver.getRefs());

        String hash = builderFactory.builder().command("index-pack")
                .argument(pack.toAbsolutePath().toString())
                .workDir(pack.getParent())
                .build(new StringStdioHandler())
                .run();
        assertEquals("539cf1050a0da7a7976dadd88a2671d15ac4e3a3", hash);
    }

    @Test
    public void testUploadPackAdvertiseRefsStatelessRpc() throws Exception {
        Path pack = tempDir.newFile("test.pack");
        RpcUploadPackRequest request = RpcUploadPackRequest.newBuilder()
                .setAdvertiseRefs(true)
                .setRepository(repository.getId())
                .setStatelessRpc(true)
                .setStrict(true)
                .build();

        UploadPackResponseObserver responseObserver = new UploadPackResponseObserver(request, null, pack);
        stub.uploadPack(responseObserver);

        //Verify the ref advertisement was served
        RpcHostingStreamResult result = responseObserver.asResult();
        assertEquals(0, result.getExitCode());
        assertEquals(0, result.getProtocolVersion());
        assertSame(RpcPackCacheResult.PACK_CACHE_NONE, result.getCacheResult()); //No ref advertisement caching
        assertSame(RpcHostingOperationType.HOSTING_OP_REF_ADVERTISEMENT, result.getOperationType());
        assertRefs(responseObserver.refs);

        //No pack data should have been written
        assertEquals(0L, Files.size(pack));
    }

    @Test
    public void testUploadPackErrorsOnUnknownWant() throws Exception {
        String arguments = "0090want 58b8128a3a3aed8add0eb86246590783ec9ed785 multi_ack_detailed side-band-64k thin-pack ofs-delta deepen-since deepen-not agent=git/2.28.0\n" +
                "00000009done\n";
        Path pack = tempDir.newFile("never-written.pack");
        RpcUploadPackRequest request = RpcUploadPackRequest.newBuilder()
                .setRepository(repository.getId())
                .build();

        UploadPackResponseObserver responseObserver = new UploadPackResponseObserver(request, arguments, pack);
        stub.uploadPack(responseObserver);

        RpcHostingStreamResult result = responseObserver.asResult();

        assertNotEquals(0, result.getExitCode());
        assertThat(responseObserver.getStderr(), containsString("fatal: git upload-pack: not our ref"));
    }

    private static void assertRefs(List<String> refs) {
        assertThat(refs, contains(
                containsString("symref=HEAD:refs/heads/master"),
                containsString("0048d6edcbf924697ab811a867421dab60d954ccad99 refs/heads/basic_branching"),
                containsString("00496053a1eaa1c009dd11092d09a72f3c41af1b59ad refs/heads/branch_mod_merge"),
                containsString("003f0a943a29376f2336b78312d99e65da17048951db refs/heads/master"),
                containsString("004c2e10dd2d1d5eea9291b296e78312e8a703964a95 refs/heads/out_of_order_branch"),
                containsString("004f80aa5be5ecf39660f798858482254f7a2ae9110e refs/tags/backdated_annotated_tag"),
                containsString("005257150c54c38d6570b2fd5e6d6cfc19476de44e84 refs/tags/backdated_annotated_tag^{}"),
                containsString("004b9b4e781dea0769888fe270e06ad76675f73851b1 refs/tags/retagged_signed_tag"),
                containsString("004e0a943a29376f2336b78312d99e65da17048951db refs/tags/retagged_signed_tag^{}"),
                containsString("004212ebe2a58367347cd39f19f5a72f3cbec7b8f9a9 refs/tags/signed_tag"),
                containsString("00450a943a29376f2336b78312d99e65da17048951db refs/tags/signed_tag^{}")));
    }

    private static class UploadPackResponseObserver
            implements ClientResponseObserver<RpcUploadPackFragment, RpcHostingReadStreamFragment> {

        private final String arguments;
        private final Path pack;
        private final List<String> refs;
        private final RpcUploadPackRequest request;
        private final CompletableFuture<RpcHostingStreamResult> resultFuture;

        private boolean inRefsSection;
        private GitPacketStream packetStream;
        private volatile ClientCallStreamObserver<RpcUploadPackFragment> requestStream;
        private RpcHostingStreamResult result;
        private ByteArrayOutputStream stderr;
        private OutputStream writer;

        UploadPackResponseObserver(RpcUploadPackRequest request, String arguments, Path pack) {
            this.arguments = arguments;
            this.pack = pack;
            this.request = request;

            refs = new ArrayList<>();
            resultFuture = new CompletableFuture<>();
        }

        @Override
        public void beforeStart(ClientCallStreamObserver<RpcUploadPackFragment> 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:
                    break;
                case STDERR:
                    onStderr(value.getStderr());
                    break;
                case STDOUT:
                    onStdout(value.getStdout());
                    break;
            }
        }

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

        List<String> getRefs() {
            return refs;
        }

        String getStderr() {
            return stderr.toString();
        }

        private void maybeAddRef(String line) {
            //Add any line that comes prior to the flush packet (which indicates the end of the refs)
            inRefsSection = inRefsSection && !line.equals(FLUSH_PACKET);
            if (inRefsSection) {
                refs.add(line);
            }
        }

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

        private void onStderr(RpcStreamChunk chunk) {
            try {
                if (stderr == null) {
                    stderr = new ByteArrayOutputStream();
                }
                chunk.getData().writeTo(stderr);
            } catch (IOException e) {
                throw new UncheckedIOException(e);
            }
        }

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

                    //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(false, 1024);
                    packetStream.addListener(new PackfileListener(writer, this::maybeAddRef, (buf, closed) -> { }));

                    inRefsSection = true;
                }
                for (ByteBuffer buffer : chunk.getData().asReadOnlyByteBufferList()) {
                    packetStream.append(buffer);
                }

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