package com.atlassian.bitbucket.mesh.git;

import com.atlassian.bitbucket.mesh.execution.GrpcAsyncExecutor;
import com.atlassian.bitbucket.mesh.git.batch.BatchChangesParameters;
import com.atlassian.bitbucket.mesh.git.batch.BatchManager;
import com.atlassian.bitbucket.mesh.grpc.BackoffStreamObserver;
import com.atlassian.bitbucket.mesh.repository.Repository;
import com.atlassian.bitbucket.mesh.repository.RepositoryManager;
import com.atlassian.bitbucket.mesh.rpc.v1.BatchServiceGrpc;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcBatchChangesRequest;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcBatchChangesResponseFragment;
import io.grpc.stub.StreamObserver;

import static com.atlassian.bitbucket.mesh.rpc.util.ByteStringUtils.emptyToNull;

public class GrpcBatchService extends BatchServiceGrpc.BatchServiceImplBase {

    private final BatchManager batchManager;
    private final GrpcAsyncExecutor executor;
    private final RepositoryManager repositoryManager;

    public GrpcBatchService(BatchManager batchManager, GrpcAsyncExecutor executor,
                            RepositoryManager repositoryManager) {
        this.batchManager = batchManager;
        this.executor = executor;
        this.repositoryManager = repositoryManager;
    }

    @Override
    public void changes(RpcBatchChangesRequest request, StreamObserver<RpcBatchChangesResponseFragment> responseObserver) {
        Repository repository = repositoryManager.getById(request.getRepository());
        BatchChangesParameters parameters = new BatchChangesParameters.Builder(repository, request.getTimeouts())
                .sinceCommitId(emptyToNull(request.getSinceCommitish()))
                .untilCommitId(request.getUntilCommitish().toStringUtf8())
                .build();

        ObserverBatchChangeCallback callback =
                new ObserverBatchChangeCallback(new BackoffStreamObserver<>(responseObserver));
        executor.execute(responseObserver, () -> {
            batchManager.changes(parameters, callback);

            //If we make it here, no exceptions were thrown trying to stream the data. Call onEnd on the
            //callback to complete the request (and flush any pending data). This isn't in a finally block
            //because if an exception is thrown above this shouldn't be called
            callback.onEnd();
        });
    }
}
