package com.atlassian.bitbucket.mesh.git;

import com.atlassian.bitbucket.mesh.io.IoSupplier;
import com.atlassian.bitbucket.mesh.repository.Repository;
import com.atlassian.bitbucket.mesh.repository.RepositoryManager;
import com.atlassian.bitbucket.mesh.rpc.v1.*;
import io.grpc.stub.ServerCallStreamObserver;
import io.grpc.stub.StreamObserver;

import javax.annotation.Nonnull;
import java.io.InputStream;

public class GrpcLfsService extends LfsServiceGrpc.LfsServiceImplBase {

    private final LfsManager lfsManager;
    private final RepositoryManager repositoryManager;

    public GrpcLfsService(LfsManager lfsManager, RepositoryManager repositoryManager) {
        this.lfsManager = lfsManager;
        this.repositoryManager = repositoryManager;
    }

    @Override
    public void canStore(RpcLfsCanStoreRequest request, StreamObserver<RpcLfsCanStoreResponse> responseObserver) {
        Repository repository = repositoryManager.getById(request.getRepository());
        responseObserver.onNext(RpcLfsCanStoreResponse.newBuilder()
                .setCanStore(lfsManager.canStore(repository, request.getSize()))
                .build());
        responseObserver.onCompleted();
    }

    @Override
    public void objectExists(RpcLfsObjectExistsRequest request, StreamObserver<RpcLfsObjectExistsResponse> responseObserver) {
        Repository repository = repositoryManager.getById(request.getRepository());
        responseObserver.onNext(RpcLfsObjectExistsResponse.newBuilder()
                .setExists(lfsManager.objectExists(repository, request.getOid()))
                .build());
        responseObserver.onCompleted();
    }

    @Override
    public void readObject(RpcLfsReadObjectRequest request, StreamObserver<RpcStreamChunk> responseObserver) {
        if (!(responseObserver instanceof ServerCallStreamObserver)) {
            throw new IllegalArgumentException("A ServerCallStreamObserver is required");
        }

        Repository repository = repositoryManager.getById(request.getRepository());
        IoSupplier<InputStream> inputSupplier;
        if (request.hasRange()) {
            RpcLfsByteRange range = request.getRange();
            inputSupplier = lfsManager.readObject(repository, request.getOid(), range.getStart(), range.getEnd());
        } else {
            inputSupplier = lfsManager.readObject(repository, request.getOid());
        }
        ServerCallStreamObserver<RpcStreamChunk> observer = (ServerCallStreamObserver<RpcStreamChunk>) responseObserver;
        Runnable handler = new LfsReadObjectHandler(inputSupplier, observer);
        observer.setOnReadyHandler(handler);
        observer.setOnCancelHandler(handler);
    }

    @Nonnull
    @Override
    public StreamObserver<RpcLfsWriteObjectFragment> writeObject(StreamObserver<RpcLfsWriteObjectResponse> responseObserver) {
        return new LfsWriteObjectRequestObserver(responseObserver, lfsManager, repositoryManager);
    }
}