package com.atlassian.bitbucket.mesh.git;

import com.atlassian.bitbucket.mesh.AbstractStatusException;
import com.atlassian.bitbucket.mesh.git.exception.*;
import com.atlassian.bitbucket.mesh.repository.Repository;
import com.atlassian.bitbucket.mesh.repository.RepositoryManager;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcLfsWriteObjectFragment;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcLfsWriteObjectRequest;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcLfsWriteObjectResponse;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcStreamChunk;
import io.grpc.Status;
import io.grpc.stub.StreamObserver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import java.nio.ByteBuffer;

public class LfsWriteObjectRequestObserver implements StreamObserver<RpcLfsWriteObjectFragment> {

    private static final Logger log = LoggerFactory.getLogger(LfsWriteObjectRequestObserver.class);

    private final LfsManager lfsManager;
    private final RepositoryManager repositoryManager;
    private final StreamObserver<RpcLfsWriteObjectResponse> responseObserver;

    private volatile boolean completed;
    private volatile LfsObjectWriter writer;

    public LfsWriteObjectRequestObserver(@Nonnull StreamObserver<RpcLfsWriteObjectResponse> responseObserver,
                                         @Nonnull LfsManager lfsManager, @Nonnull RepositoryManager repositoryManager) {
        this.responseObserver = responseObserver;
        this.lfsManager = lfsManager;
        this.repositoryManager = repositoryManager;
    }

    @Override
    public void onNext(RpcLfsWriteObjectFragment fragment) {
        try {
            switch (fragment.getFragmentOneofCase()) {
                case REQUEST:
                    handleRequest(fragment.getRequest());
                    break;
                case CONTENT:
                    handleContent(fragment.getContent());
                    break;
                default:
                    throw new InvalidArgumentException("[" + fragment.getFragmentOneofCase() + "] is not supported");
            }
        } catch (RuntimeException e) {
            if (e instanceof AbstractStatusException) {
                e = ((AbstractStatusException) e).toStatusException();
            } else {
                e = new LfsStorageException(e).toStatusException();
            }
            sendError(e);
        }
    }

    @Override
    public void onError(Throwable t) {
        if (Status.fromThrowable(t).getCode() == Status.Code.CANCELLED) {
            // Log cancellation more quietly; it's not an error for the client to cancel
            log.debug("The client has canceled the request");
        } else {
            log.error("Canceling the process; the client has failed", t);
        }

        if (writer != null) {
            writer.abort();
        }

        // If we get here from our own onError callback in onNext below, this won't do anything. But if
        // we get here because the client canceled the request, or had some other failure, then we need
        // to complete the server-side observer to prevent the RPC from lingering until its deadline
        maybeComplete();
    }

    @Override
    public void onCompleted() {
        try {
            if (writer != null) {
                writer.close();
            }

            maybeComplete();
        } catch (RuntimeException e) {
            if (e instanceof AbstractStatusException) {
                e = ((AbstractStatusException) e).toStatusException();
            }
            sendError(e);
        }
    }

    private void handleContent(RpcStreamChunk content) {
        if (writer == null) {
            throw new InvalidArgumentException("An RpcLfsWriteObjectRequest must be sent before content");
        }
        for (ByteBuffer buffer : content.getData().asReadOnlyByteBufferList()) {
            writer.write(buffer);
        }
    }

    private void handleRequest(RpcLfsWriteObjectRequest request) {
        if (writer != null) {
            throw new InvalidArgumentException("Only a single RpcLfsWriteObjectRequest may be sent");
        }

        Repository repository = repositoryManager.getById(request.getRepository());
        if (repository.isReadOnly()) {
            throw new RepositoryReadOnlyException();
        }

        writer = lfsManager.writeObject(repository, request.getOid());
    }

    private void maybeComplete() {
        if (!completed) {
            completed = true;
            responseObserver.onNext(RpcLfsWriteObjectResponse.getDefaultInstance());
            responseObserver.onCompleted();
        }
    }

    private void sendError(RuntimeException e) {
        completed = true;   // If we send the error there is no need to call onComplete()
        responseObserver.onError(e);
    }
}
