package com.atlassian.bitbucket.mesh.git;

import com.atlassian.bitbucket.mesh.AbstractGrpcServiceTest;
import com.atlassian.bitbucket.mesh.git.builder.GitProcessBuilderFactory;
import com.atlassian.bitbucket.mesh.io.MoreFiles;
import com.atlassian.bitbucket.mesh.junit.LogListener;
import com.atlassian.bitbucket.mesh.process.handler.InputStdioHandler;
import com.atlassian.bitbucket.mesh.process.handler.StringStdioHandler;
import com.atlassian.bitbucket.mesh.repository.*;
import com.atlassian.bitbucket.mesh.rpc.v1.*;
import com.atlassian.bitbucket.mesh.test.ExecutionContextRule;
import com.atlassian.bitbucket.mesh.test.HasGitError;
import com.atlassian.bitbucket.mesh.test.annotation.DirtiesRepo;
import com.atlassian.bitbucket.mesh.test.annotation.Repo;
import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableList;
import com.google.common.io.ByteStreams;
import com.google.protobuf.ByteString;
import com.google.protobuf.Timestamp;
import com.google.protobuf.UnsafeByteOperations;
import io.grpc.ManagedChannel;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.grpc.stub.StreamObserver;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.FileSystemResource;
import org.springframework.core.io.Resource;

import javax.annotation.Nonnull;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
import java.nio.channels.Channels;
import java.nio.channels.WritableByteChannel;
import java.nio.file.FileVisitResult;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.SimpleFileVisitor;
import java.nio.file.attribute.BasicFileAttributes;
import java.util.Collection;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Stream;

import static com.atlassian.bitbucket.mesh.rpc.util.ByteStringUtils.toStrings;
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Collections.emptySet;
import static java.util.Collections.singleton;

public class GrpcGitDataServiceTest extends AbstractGrpcServiceTest {

    @Rule
    public final LogListener logListener = new LogListener("com.atlassian");

    @Autowired
    @Rule
    public ExecutionContextRule executionContextRule;

    @Autowired
    private GitAgent agent;
    @Repo
    private BitbucketServerRepository bitbucketServerRepository;
    @Autowired
    private GitProcessBuilderFactory builderFactory;
    @Autowired
    private ManagedChannel managedChannel;
    @Repo
    private ManagedRepository managedRepository;
    @Repo(GitRepositories.MERGE)
    private Repository merge;
    @Repo
    private ManagedRepository repository;
    @Autowired
    private RepositoryManager repositoryManager;

    @Before
    public void setUp() {
        //Remove any dirty marker that exists for the repository so we can verify whether uploading
        //the objects triggers the repository to be marked dirty.
        clearDirtyMarker(repository);
    }

    @DirtiesRepo("repository")
    @Test
    public void testUploadFile() throws InterruptedException, TimeoutException, IOException {
        doTestUploadFile(repository);
    }

    @Test
    public void testUploadFileLooseRefMarksRepoDirty() throws TimeoutException, InterruptedException, IOException {
        UploadResponseObserver<RpcUploadFileRequest, RpcUploadFileResponse> responseObserver =
                new UploadResponseObserver<>();
        StreamObserver<RpcUploadFileFragment> requestObserver = stub().uploadFile(responseObserver);

        byte[] content = "53979b3304bece75a3852a3d250dbaed64d45430".getBytes(UTF_8);
        //Send the upload file request
        requestObserver.onNext(RpcUploadFileFragment.newBuilder()
                .setRequest(RpcUploadFileRequest.newBuilder()
                        .setRepository(repository.getId())
                        .setFile(RpcOsFile.newBuilder()
                                .setLastModified(Timestamp.newBuilder().setSeconds(1623659279L))
                                .setPath(ByteString.copyFromUtf8("heads/a-branch"))
                                .setSize(content.length))
                        .setType(RpcUploadType.UPLOAD_TYPE_GIT_REFS))
                .build());

        //Wait for the server to tell us whether it wants the file or not
        RpcUploadFileResponse response = responseObserver.awaitResponse(5, TimeUnit.SECONDS);
        assertTrue(response.getWant());

        //Stream the content to the server
        requestObserver.onNext(RpcUploadFileFragment.newBuilder()
                .setChunk(RpcStreamChunk.newBuilder()
                        .setData(UnsafeByteOperations.unsafeWrap(content))
                        .setClosed(true))
                .build());
        requestObserver.onCompleted();

        responseObserver.awaitCompleted(5, TimeUnit.SECONDS);

        //Verify that the file has been stored on disk
        Path expectedPath = repository.getPath().resolve("refs/heads/a-branch");
        assertTrue(expectedPath + " is not a regular file", Files.isRegularFile(expectedPath));
        assertEquals("53979b3304bece75a3852a3d250dbaed64d45430", MoreFiles.toString(expectedPath));
        //Verify that the repository has been marked dirty
        assertMarkedDirty(true, repository);
    }

    @Test
    public void testUploadFilePackedRefsMarksRepoDirty() throws TimeoutException, InterruptedException, IOException {
        UploadResponseObserver<RpcUploadFileRequest, RpcUploadFileResponse> responseObserver =
                new UploadResponseObserver<>();
        StreamObserver<RpcUploadFileFragment> requestObserver = stub().uploadFile(responseObserver);

        String content = "# pack-refs with: peeled fully-peeled sorted \n" +
                "d6edcbf924697ab811a867421dab60d954ccad99 refs/remotes/origin/basic_branching\n" +
                "6053a1eaa1c009dd11092d09a72f3c41af1b59ad refs/remotes/origin/branch_mod_merge\n" +
                "0a943a29376f2336b78312d99e65da17048951db refs/remotes/origin/master\n" +
                "2e10dd2d1d5eea9291b296e78312e8a703964a95 refs/remotes/origin/out_of_order_branch\n" +
                "80aa5be5ecf39660f798858482254f7a2ae9110e refs/tags/backdated_annotated_tag\n" +
                "^57150c54c38d6570b2fd5e6d6cfc19476de44e84\n" +
                "9b4e781dea0769888fe270e06ad76675f73851b1 refs/tags/retagged_signed_tag\n" +
                "^0a943a29376f2336b78312d99e65da17048951db\n" +
                "12ebe2a58367347cd39f19f5a72f3cbec7b8f9a9 refs/tags/signed_tag\n" +
                "^0a943a29376f2336b78312d99e65da17048951db\n";
        byte[] contentBytes = content.getBytes(UTF_8);

        //Send the upload file request
        requestObserver.onNext(RpcUploadFileFragment.newBuilder()
                .setRequest(RpcUploadFileRequest.newBuilder()
                        .setRepository(repository.getId())
                        .setFile(RpcOsFile.newBuilder()
                                .setLastModified(Timestamp.newBuilder().setSeconds(1623659279L))
                                .setPath(ByteString.copyFromUtf8("packed-refs"))
                                .setSize(contentBytes.length))
                        .setType(RpcUploadType.UPLOAD_TYPE_GIT_ROOT))
                .build());

        //Wait for the server to tell us whether it wants the file or not
        RpcUploadFileResponse response = responseObserver.awaitResponse(5, TimeUnit.SECONDS);
        assertTrue(response.getWant());

        //Stream the content to the server
        requestObserver.onNext(RpcUploadFileFragment.newBuilder()
                .setChunk(RpcStreamChunk.newBuilder()
                        .setData(UnsafeByteOperations.unsafeWrap(contentBytes))
                        .setClosed(true))
                .build());
        requestObserver.onCompleted();

        responseObserver.awaitCompleted(5, TimeUnit.SECONDS);

        //Verify that the file has been stored on disk
        Path expectedPath = repository.getPath().resolve("packed-refs");
        assertTrue(expectedPath + " is not a regular file", Files.isRegularFile(expectedPath));
        assertEquals(content, MoreFiles.toString(expectedPath));
        //Verify that the repository has been marked dirty
        assertMarkedDirty(true, repository);
    }

    @Test
    public void testUploadFileThatAlreadyExists() throws InterruptedException, TimeoutException {
        UploadResponseObserver<RpcUploadFileRequest, RpcUploadFileResponse> responseObserver =
                new UploadResponseObserver<>();
        StreamObserver<RpcUploadFileFragment> requestObserver = stub().uploadFile(responseObserver);

        RpcOsFile objectFile = getAnyObjectFile(repository);
        assertNotNull(objectFile);

        //Send the upload file request
        requestObserver.onNext(RpcUploadFileFragment.newBuilder()
                .setRequest(RpcUploadFileRequest.newBuilder()
                        .setRepository(repository.getId())
                        .setFile(objectFile)
                        .setType(RpcUploadType.UPLOAD_TYPE_GIT_OBJECTS))
                .build());

        RpcUploadFileResponse response = responseObserver.awaitResponse(5, TimeUnit.SECONDS);
        assertFalse(response.getWant());

        //Tell the server we're done and wait for the server to wrap up too
        requestObserver.onCompleted();
        responseObserver.awaitCompleted(5, TimeUnit.SECONDS);

        assertMarkedDirty(false, repository);
    }

    @Test
    public void testUploadFileToEmptyRepository() throws InterruptedException, TimeoutException, IOException {
        Repository repository = repositoryManager.create(new CreateRepositoryParameters.Builder(
                ManagedRepository.formatId(0, "aaaaabbbbbcccccddddd", "12376")).build());
        try {
            clearDirtyMarker(repository);
            doTestUploadFile(repository);
        } finally {
            repositoryManager.delete(new DeleteRepositoryParameters.Builder(repository).build());
        }
    }

    @DirtiesRepo("repository")
    @Test
    public void testUploadFileTranslatesDarkReflog() throws InterruptedException, TimeoutException, IOException {
        UploadResponseObserver<RpcUploadFileRequest, RpcUploadFileResponse> responseObserver =
                new UploadResponseObserver<>();
        StreamObserver<RpcUploadFileFragment> requestObserver = stub().uploadFile(responseObserver);

        byte[] content = "aec04128b0e9fead9d52e6d09b24675035281d22".getBytes(UTF_8);
        //Send the upload file request
        requestObserver.onNext(RpcUploadFileFragment.newBuilder()
                .setRequest(RpcUploadFileRequest.newBuilder()
                        .setRepository(repository.getId())
                        .setFile(RpcOsFile.newBuilder()
                                .setLastModified(Timestamp.newBuilder().setSeconds(1623659279L))
                                .setPath(ByteString.copyFromUtf8("stash-refs/pull-requests/12/from"))
                                .setSize(content.length))
                        .setType(RpcUploadType.UPLOAD_TYPE_GIT_LOG))
                .build());

        //Wait for the server to tell us whether it wants the file or not
        RpcUploadFileResponse response = responseObserver.awaitResponse(5, TimeUnit.SECONDS);
        assertTrue(response.getWant());

        //Stream the content to the server
        requestObserver.onNext(RpcUploadFileFragment.newBuilder()
                .setChunk(RpcStreamChunk.newBuilder()
                        .setData(UnsafeByteOperations.unsafeWrap(content))
                        .setClosed(true))
                .build());
        requestObserver.onCompleted();

        responseObserver.awaitCompleted(5, TimeUnit.SECONDS);

        //Verify that the file has been stored on disk
        Path expectedPath = repository.getPath().resolve("logs/dark-refs/pull-requests/12/from");
        assertTrue(expectedPath + " is not a regular file", Files.isRegularFile(expectedPath));
        assertEquals("aec04128b0e9fead9d52e6d09b24675035281d22", MoreFiles.toString(expectedPath));

        assertMarkedDirty(true, repository);
    }

    @DirtiesRepo("repository")
    @Test
    public void testUploadFiles() throws IOException, InterruptedException, TimeoutException {
        doTestUploadFiles(repository);
    }

    @Test
    public void testUploadFilesToEmptyRepository() throws IOException, InterruptedException, TimeoutException {
        Repository repository = repositoryManager.create(new CreateRepositoryParameters.Builder(
                ManagedRepository.formatId(0, "bbbbbcccccdddddeeeee", "68234")).build());
        try {
            clearDirtyMarker(repository);
            doTestUploadFiles(repository);
        } finally {
            repositoryManager.delete(new DeleteRepositoryParameters.Builder(repository).build());
        }
    }

    @DirtiesRepo("repository")
    @Test
    public void testUploadLargeFile() throws Exception {
        //Upload a file using large segments that will be split up into separate Netty buffers on the server side to
        //truly exercise the marshalling optimisation we have in StreamChunkParser
        UploadResponseObserver<RpcUploadFileRequest, RpcUploadFileResponse> responseObserver =
                new UploadResponseObserver<>();
        StreamObserver<RpcUploadFileFragment> requestObserver = stub().uploadFile(responseObserver);

        byte[] contents = new byte[1024 * 1024]; // 1Mb
        new Random(12345347892L).nextBytes(contents); // generate random contents

        //Send the upload file request
        requestObserver.onNext(RpcUploadFileFragment.newBuilder()
                .setRequest(RpcUploadFileRequest.newBuilder()
                        .setRepository(repository.getId())
                        .setFile(RpcOsFile.newBuilder()
                                .setLastModified(Timestamp.newBuilder().setSeconds(1623659279L))
                                .setPath(ByteString.copyFromUtf8("ae/c04128b0e9fead9d52e6d09b24675035281d22"))
                                .setSize(contents.length))
                        .setType(RpcUploadType.UPLOAD_TYPE_GIT_OBJECTS))
                .build());

        //Wait for the server to tell us whether it wants the file or not
        RpcUploadFileResponse response = responseObserver.awaitResponse(5, TimeUnit.SECONDS);
        assertTrue(response.getWant());

        //Stream the file to the server in 2 512kb chunks
        int firstChunkSize = contents.length / 2;
        requestObserver.onNext(RpcUploadFileFragment.newBuilder()
                .setChunk(RpcStreamChunk.newBuilder()
                        .setData(ByteString.copyFrom(contents, 0, firstChunkSize)))
                .build());
        requestObserver.onNext(RpcUploadFileFragment.newBuilder()
                .setChunk(RpcStreamChunk.newBuilder()
                        .setData(ByteString.copyFrom(contents, firstChunkSize, contents.length - firstChunkSize))
                        .setClosed(true))
                .build());
        requestObserver.onCompleted();
        responseObserver.awaitCompleted(5, TimeUnit.SECONDS);

        //Verify that the file has been stored on disk
        Path expectedPath = repository.getPath().resolve("objects/ae/c04128b0e9fead9d52e6d09b24675035281d22");
        assertTrue(expectedPath + " is not a regular file", Files.isRegularFile(expectedPath));

        //Verify the attributes
        BasicFileAttributes attributes = Files.readAttributes(expectedPath, BasicFileAttributes.class);
        assertEquals(contents.length, attributes.size());
        assertEquals(1623659279000L, attributes.lastModifiedTime().toMillis());

        //Verify the contents
        assertArrayEquals(contents, getResourceBytes(new FileSystemResource(expectedPath)));

        //The file has been uploaded to the repository itself (not shared-objects), so it shouldn't be marked dirty
        assertMarkedDirty(false, repository);
    }

    @DirtiesRepo("repository")
    @Test
    public void testUploadPackFileBitbucketServerRepository() throws InterruptedException {
        doTestUploadPack(bitbucketServerRepository);
    }

    @Test
    public void testUploadPackFileEmptyRepository() throws InterruptedException {
        Repository repository = repositoryManager.create(new CreateRepositoryParameters.Builder(
                ManagedRepository.formatId(0, "bbbbbcccccdddddeeeee", "68234")).build());
        try {
            clearDirtyMarker(repository);
            doTestUploadPack(repository);
        } finally {
            repositoryManager.delete(new DeleteRepositoryParameters.Builder(repository).build());
        }
    }

    @Test
    public void testUploadPackFileManagedRepository() throws InterruptedException {
        doTestUploadPack(managedRepository);
    }

    @DirtiesRepo("merge")
    @Test
    public void testUploadPackFileThrowsWhenPackDependsOnMissingObjects() throws IOException {
        //Create a pack with just the contents of the initial commit of the merge repository
        byte[] packContent = getPackFileContents(repository,
                singleton("042fb0cc198e7320b59a025689b45a2cb52f4684"),
                singleton("53979b3304bece75a3852a3d250dbaed64d45430"));

        UploadPackFileResponseObserver responseObserver = new UploadPackFileResponseObserver();
        StreamObserver<RpcUploadPackFileFragment> observer = stub().uploadPackFile(responseObserver);

        //Send the request and the valid pack
        observer.onNext(RpcUploadPackFileFragment.newBuilder()
                .setRequest(RpcUploadPackFileRequest.newBuilder()
                        .setRepository(merge.getId()))
                .build());
        observer.onNext(RpcUploadPackFileFragment.newBuilder()
                .setChunk(RpcStreamChunk.newBuilder()
                        .setData(ByteString.copyFrom(packContent))
                        .setClosed(true)
                        .build())
                .build());
        observer.onCompleted();

        //Wait for the upload to complete; this should throw
        StatusRuntimeException thrown = assertThrows(StatusRuntimeException.class, responseObserver::awaitCompleted);
        assertThat(thrown, HasGitError.invalidPack(
                RpcInvalidPack.Reason.MISSING_OBJECT,
                "did not receive expected object 4c54863857092fad52e494e1b3db94fe4e3b136e"));

        //Validate that no partial pack file lingers in the pack dir
        try (Stream<Path> contents = Files.list(merge.getPath().resolve("objects/pack"))) {
            assertEquals(0, contents.filter(path -> path.endsWith(".pack")).count());
        }

        logListener.assertWarn("Invalid pack: MISSING_OBJECT " +
                "(did not receive expected object 4c54863857092fad52e494e1b3db94fe4e3b136e)");
    }

    @DirtiesRepo("repository")
    @Test
    public void testUploadPackFileThrowsWhenTruncated() throws IOException {
        //Create a pack with just the contents of the initial commit of the merge repository
        byte[] packContent = getPackFileContents(
                merge, singleton("017caf31eca7c46eb3d1800fcac431cfa7147a01"), emptySet());

        UploadPackFileResponseObserver responseObserver = new UploadPackFileResponseObserver();
        StreamObserver<RpcUploadPackFileFragment> observer = stub().uploadPackFile(responseObserver);

        //First send the request
        observer.onNext(RpcUploadPackFileFragment.newBuilder()
                .setRequest(RpcUploadPackFileRequest.newBuilder()
                        .setRepository(repository.getId()))
                .build());

        //Send a partial pack and complete the stream
        int firstChunkSize = packContent.length / 2;
        observer.onNext(RpcUploadPackFileFragment.newBuilder()
                .setChunk(RpcStreamChunk.newBuilder()
                        .setData(ByteString.copyFrom(packContent, 0, firstChunkSize))
                        .setClosed(true)
                        .build())
                .build());
        observer.onCompleted();

        //Wait for the upload to complete; this should throw
        StatusRuntimeException thrown = assertThrows(StatusRuntimeException.class, responseObserver::awaitCompleted);
        assertThat(thrown, HasGitError.invalidPack(RpcInvalidPack.Reason.EARLY_EOF));
        assertEquals(Status.INVALID_ARGUMENT.getCode(), thrown.getStatus().getCode());

        //Validate that no partial pack file lingers in the pack dir
        try (Stream<Path> contents = Files.list(repository.getPath().resolve("objects/pack"))) {
            assertEquals(0, contents.filter(path -> path.endsWith(".pack")).count());
        }
        logListener.assertWarn("Invalid pack: EARLY_EOF");
    }

    private static void assertMarkedDirty(boolean expected, Repository repository) {
        if (repository instanceof ManagedRepository) {
            ManagedRepository managed = (ManagedRepository) repository;
            assertEquals("The repository was " + (expected ? "not" : "") + " marked dirty",
                    expected, managed.isDirty());
        }
    }

    private static void clearDirtyMarker(Repository repository) {
        if (repository instanceof ManagedRepository) {
            //Remove any dirty marker that exists for the repository so we can verify whether uploading
            //the objects triggers the repository to be marked dirty.
            ManagedRepository managed = (ManagedRepository) repository;
            MoreFiles.deleteQuietly(managed.getSharedObjects().getDirtyMarker(managed));
        }
    }

    private static RpcOsFile getAnyObjectFile(Repository repository) {
        try {
            Path objectsDir = repository.getPath().resolve("objects");
            AtomicReference<RpcOsFile> result = new AtomicReference<>();
            Files.walkFileTree(objectsDir, new SimpleFileVisitor<Path>() {

                @Override
                public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) {
                    result.set(RpcOsFile.newBuilder()
                            .setLastModified(Timestamp.newBuilder().setSeconds(1623659279L))
                            .setPath(ByteString.copyFromUtf8(objectsDir.relativize(file).toString()))
                            .setSize(attrs.size())
                            .build());
                    return FileVisitResult.TERMINATE;
                }
            });
            return result.get();
        } catch (IOException e) {
            return null;
        }
    }

    private static byte[] getResourceBytes(Resource resource) throws IOException {
        try (InputStream is = resource.getInputStream()) {
            return ByteStreams.toByteArray(is);
        }
    }

    private void doTestUploadFile(Repository repository) throws InterruptedException, TimeoutException, IOException {
        boolean wasEmpty = !agent.hasObjects(repository);
        UploadResponseObserver<RpcUploadFileRequest, RpcUploadFileResponse> responseObserver =
                new UploadResponseObserver<>();
        StreamObserver<RpcUploadFileFragment> requestObserver = stub().uploadFile(responseObserver);

        //Send the upload file request
        requestObserver.onNext(RpcUploadFileFragment.newBuilder()
                .setRequest(RpcUploadFileRequest.newBuilder()
                        .setRepository(repository.getId())
                        .setFile(RpcOsFile.newBuilder()
                                .setLastModified(Timestamp.newBuilder().setSeconds(1623659279L))
                                .setPath(ByteString.copyFromUtf8("ae/c04128b0e9fead9d52e6d09b24675035281d22"))
                                .setSize(36L))
                        .setType(RpcUploadType.UPLOAD_TYPE_GIT_OBJECTS))
                .build());

        //Wait for the server to tell us whether it wants the file or not
        RpcUploadFileResponse response = responseObserver.awaitResponse(5, TimeUnit.SECONDS);
        assertTrue(response.getWant());

        byte[] bytes = getResourceBytes(
                new ClassPathResource("/coordinator-service/aec04128b0e9fead9d52e6d09b24675035281d22"));

        //Stream the file to the server in 2 chunks
        requestObserver.onNext(RpcUploadFileFragment.newBuilder()
                .setChunk(RpcStreamChunk.newBuilder()
                        .setData(ByteString.copyFrom(bytes, 0, 10)))
                .build());
        requestObserver.onNext(RpcUploadFileFragment.newBuilder()
                .setChunk(RpcStreamChunk.newBuilder()
                        .setData(ByteString.copyFrom(bytes, 10, bytes.length - 10))
                        .setClosed(true))
                .build());
        requestObserver.onCompleted();

        responseObserver.awaitCompleted(5, TimeUnit.SECONDS);

        //Verify that the file has been stored on disk
        Path expectedPath = getDestinationObjectsDir(repository).resolve("ae/c04128b0e9fead9d52e6d09b24675035281d22");
        assertTrue(expectedPath + " is not a regular file", Files.isRegularFile(expectedPath));

        //Verify the attributes
        BasicFileAttributes attributes = Files.readAttributes(expectedPath, BasicFileAttributes.class);
        assertEquals(36L, attributes.size());
        assertEquals(1623659279000L, attributes.lastModifiedTime().toMillis());

        //Verify the contents
        assertArrayEquals(bytes, getResourceBytes(new FileSystemResource(expectedPath)));

        assertMarkedDirty(wasEmpty, repository);
    }

    private void doTestUploadFiles(Repository repository) throws IOException, InterruptedException, TimeoutException {
        boolean wasEmpty = !agent.hasObjects(repository);
        UploadResponseObserver<RpcUploadFileRequest, RpcUploadFilesResponse> responseObserver =
                new UploadResponseObserver<>();
        StreamObserver<RpcUploadFilesFragment> requestObserver = stub().uploadFiles(responseObserver);

        //Send an upload files request for 2 files that do not exist and one that already exists
        RpcOsFile file1 = RpcOsFile.newBuilder()
                .setLastModified(Timestamp.newBuilder().setSeconds(1623659279L))
                .setPath(ByteString.copyFromUtf8("ae/c04128b0e9fead9d52e6d09b24675035281d22"))
                .setSize(36L)
                .build();
        RpcOsFile file2 = RpcOsFile.newBuilder()
                .setLastModified(Timestamp.newBuilder().setSeconds(1623658888L))
                .setPath(ByteString.copyFromUtf8("a9/dc9e52a2c4061c5bc5e1405d3da4d422adac6c"))
                .setSize(363L)
                .build();

        requestObserver.onNext(RpcUploadFilesFragment.newBuilder()
                .setRequest(RpcUploadFilesRequest.newBuilder()
                        .setRepository(repository.getId())
                        .addFiles(file1)
                        .addFiles(getAnyObjectFile(repository)) // already exists
                        .addFiles(file2)
                        .setType(RpcUploadType.UPLOAD_TYPE_GIT_OBJECTS))
                .build());

        //Wait for the server to tell us what files it is interested in
        RpcUploadFilesResponse response = responseObserver.awaitResponse(5, TimeUnit.SECONDS);
        assertEquals(
                ImmutableList.of("ae/c04128b0e9fead9d52e6d09b24675035281d22",
                        "a9/dc9e52a2c4061c5bc5e1405d3da4d422adac6c"),
                toStrings(response.getWantPathsList()));

        //Stream file1 to the server in 2 chunks
        byte[] bytesFile1 = getResourceBytes(
                new ClassPathResource("/coordinator-service/aec04128b0e9fead9d52e6d09b24675035281d22"));
        requestObserver.onNext(RpcUploadFilesFragment.newBuilder()
                .setFileStart(RpcUploadFileStart.newBuilder().setPath(file1.getPath()))
                .build());
        requestObserver.onNext(RpcUploadFilesFragment.newBuilder()
                .setChunk(RpcStreamChunk.newBuilder()
                        .setData(ByteString.copyFrom(bytesFile1, 0, 10)))
                .build());
        requestObserver.onNext(RpcUploadFilesFragment.newBuilder()
                .setChunk(RpcStreamChunk.newBuilder()
                        .setData(ByteString.copyFrom(bytesFile1, 10, bytesFile1.length - 10))
                        .setClosed(true))
                .build());

        //Stream file2 to the server in 3 chunks
        byte[] bytesFile2 = getResourceBytes(
                new ClassPathResource("/coordinator-service/a9dc9e52a2c4061c5bc5e1405d3da4d422adac6c"));
        requestObserver.onNext(RpcUploadFilesFragment.newBuilder()
                .setFileStart(RpcUploadFileStart.newBuilder().setPath(file2.getPath()))
                .build());
        requestObserver.onNext(RpcUploadFilesFragment.newBuilder()
                .setChunk(RpcStreamChunk.newBuilder()
                        .setData(ByteString.copyFrom(bytesFile2, 0, 10)))
                .build());
        requestObserver.onNext(RpcUploadFilesFragment.newBuilder()
                .setChunk(RpcStreamChunk.newBuilder()
                        .setData(ByteString.copyFrom(bytesFile2, 10, 10)))
                .build());
        requestObserver.onNext(RpcUploadFilesFragment.newBuilder()
                .setChunk(RpcStreamChunk.newBuilder()
                        .setData(ByteString.copyFrom(bytesFile2, 20, bytesFile2.length - 20))
                        .setClosed(true))
                .build());

        //Simulate a second batch, consisting of a single file
        RpcOsFile file3 = RpcOsFile.newBuilder()
                .setLastModified(Timestamp.newBuilder().setSeconds(1623659999L))
                .setPath(ByteString.copyFromUtf8("ff/ffffffffffffffffffffffffffffffffffffff"))
                .setSize(36L)
                .build();
        requestObserver.onNext(RpcUploadFilesFragment.newBuilder()
                .setRequest(RpcUploadFilesRequest.newBuilder()
                        .setRepository(repository.getId())
                        .addFiles(file3)
                        .setType(RpcUploadType.UPLOAD_TYPE_GIT_OBJECTS))
                .build());

        //Wait for the server to tell us what files it is interested in
        response = responseObserver.awaitResponse(5, TimeUnit.SECONDS);
        assertEquals(ImmutableList.of("ff/ffffffffffffffffffffffffffffffffffffff"),
                toStrings(response.getWantPathsList()));

        //Stream file1 contents as file3 to the server in a single chunk
        requestObserver.onNext(RpcUploadFilesFragment.newBuilder()
                .setFileStart(RpcUploadFileStart.newBuilder().setPath(file3.getPath()))
                .build());
        requestObserver.onNext(RpcUploadFilesFragment.newBuilder()
                .setChunk(RpcStreamChunk.newBuilder()
                        .setClosed(true)
                        .setData(ByteString.copyFrom(bytesFile1)))
                .build());
        requestObserver.onCompleted();

        responseObserver.awaitCompleted(5, TimeUnit.SECONDS);

        //Verify that the file1 has been stored on disk
        Path expectedPath = getDestinationObjectsDir(repository).resolve("ae/c04128b0e9fead9d52e6d09b24675035281d22");
        assertTrue(expectedPath + " is not a regular file", Files.isRegularFile(expectedPath));

        //Verify the attributes
        BasicFileAttributes attributes = Files.readAttributes(expectedPath, BasicFileAttributes.class);
        assertEquals(36L, attributes.size());
        assertEquals(1623659279000L, attributes.lastModifiedTime().toMillis());

        //Verify the contents
        assertArrayEquals(bytesFile1, getResourceBytes(new FileSystemResource(expectedPath)));

        //Verify that the file2 has been stored on disk
        expectedPath = getDestinationObjectsDir(repository).resolve("a9/dc9e52a2c4061c5bc5e1405d3da4d422adac6c");
        assertTrue(expectedPath + " is not a regular file", Files.isRegularFile(expectedPath));

        //Verify the attributes
        attributes = Files.readAttributes(expectedPath, BasicFileAttributes.class);
        assertEquals(363L, attributes.size());
        assertEquals(1623658888000L, attributes.lastModifiedTime().toMillis());

        //Verify the contents
        assertArrayEquals(bytesFile2, getResourceBytes(new FileSystemResource(expectedPath)));

        //Verify that the file3 has been stored on disk
        expectedPath = getDestinationObjectsDir(repository).resolve("ff/ffffffffffffffffffffffffffffffffffffff");
        assertTrue(expectedPath + " is not a regular file", Files.isRegularFile(expectedPath));

        //Verify the attributes
        attributes = Files.readAttributes(expectedPath, BasicFileAttributes.class);
        assertEquals(36L, attributes.size());
        assertEquals(1623659999000L, attributes.lastModifiedTime().toMillis());

        //Verify the contents
        assertArrayEquals(bytesFile1, getResourceBytes(new FileSystemResource(expectedPath)));

        assertMarkedDirty(wasEmpty, repository);
    }

    private void doTestUploadPack(Repository repository) throws InterruptedException {
        boolean wasEmpty = !agent.hasObjects(repository);

        //Create a pack with just the contents of the initial commit of the merge repository
        byte[] packContent = getPackFileContents(
                merge, singleton("017caf31eca7c46eb3d1800fcac431cfa7147a01"), emptySet());

        UploadPackFileResponseObserver responseObserver = new UploadPackFileResponseObserver();
        StreamObserver<RpcUploadPackFileFragment> observer = stub().uploadPackFile(responseObserver);

        //First send the request
        observer.onNext(RpcUploadPackFileFragment.newBuilder()
                .setRequest(RpcUploadPackFileRequest.newBuilder()
                        .setRepository(repository.getId()))
                .build());

        //Send the pack contents in multiple chunks
        int firstChunkSize = packContent.length / 2;
        observer.onNext(RpcUploadPackFileFragment.newBuilder()
                .setChunk(RpcStreamChunk.newBuilder()
                        .setData(ByteString.copyFrom(packContent, 0, firstChunkSize))
                        .build())
                .build());
        observer.onNext(RpcUploadPackFileFragment.newBuilder()
                .setChunk(RpcStreamChunk.newBuilder()
                        .setData(ByteString.copyFrom(packContent, firstChunkSize, packContent.length - firstChunkSize))
                        .setClosed(true)
                        .build())
                .build());
        observer.onCompleted();

        //Wait for the upload to complete
        responseObserver.awaitCompleted();

        //Validate that the commit now exists in repository
        String result = builderFactory.builder(repository)
                .catFile()
                .type()
                .object("017caf31eca7c46eb3d1800fcac431cfa7147a01")
                .build(new StringStdioHandler())
                .run();
        assertEquals(result, "commit");

        assertMarkedDirty(wasEmpty, repository);
    }

    private Path getDestinationObjectsDir(Repository repository) {
        if (repository instanceof ManagedRepository && !agent.hasObjects(repository)) {
            return ((ManagedRepository) repository).getSharedObjects().getPath().resolve("objects");
        }
        return repository.getPath().resolve("objects");
    }

    private byte[] getPackFileContents(Repository repository,
                                       Collection<String> includes, Collection<String> excludes) {
        StringBuilder stdin = new StringBuilder();
        includes.forEach(include -> stdin.append(include).append('\n'));
        excludes.forEach(exclude -> stdin.append('^').append(exclude).append('\n'));
        PackObjectsStdioHandler handler = new PackObjectsStdioHandler(stdin.toString());
        builderFactory.builder(repository)
                .packObjects()
                .revs(true)
                .build(handler)
                .run();
        return handler.getBytes();
    }

    private GitDataServiceGrpc.GitDataServiceStub stub() {
        return GitDataServiceGrpc.newStub(managedChannel);
    }

    private static class PackObjectsStdioHandler extends InputStdioHandler {

        private final ByteArrayOutputStream bos;
        private final WritableByteChannel out;

        PackObjectsStdioHandler(@Nonnull String stdin) {
            super(stdin);
            bos = new ByteArrayOutputStream();
            out = Channels.newChannel(bos);
        }

        public byte[] getBytes() {
            return bos.toByteArray();
        }

        @Override
        public void onStdout(@Nonnull ByteBuffer buffer, boolean closed) {
            try {
                out.write(buffer);
                if (closed) {
                    out.close();
                }
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    private static class UploadPackFileResponseObserver implements StreamObserver<RpcUploadPackFileResponse> {

        private final CompletableFuture<RpcUploadPackFileResponse> future = new CompletableFuture<>();

        private RpcUploadPackFileResponse response;

        public void awaitCompleted() throws InterruptedException {
            try {
                future.get();
            } catch (ExecutionException e) {
                Throwable cause = e.getCause();
                Throwables.throwIfUnchecked(cause);
                throw new RuntimeException(cause);
            }
        }

        @Override
        public void onCompleted() {
            future.complete(response);
        }

        @Override
        public void onError(Throwable throwable) {
            future.completeExceptionally(throwable);
        }

        @Override
        public void onNext(RpcUploadPackFileResponse response) {
            this.response = response;
        }
    }
}
