package com.atlassian.bitbucket.mesh.git;

import com.atlassian.bitbucket.mesh.AbstractStatusException;
import com.atlassian.bitbucket.mesh.git.worktree.WorkTreeWriteException;
import com.atlassian.bitbucket.mesh.repository.WorkTree;
import com.atlassian.bitbucket.mesh.repository.WorkTreeManager;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcStreamChunk;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcWriteWorkTreeFileFragment;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcWriteWorkTreeFileRequest;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcWriteWorkTreeFileResponse;
import io.grpc.Status;
import io.grpc.stub.StreamObserver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.WritableByteChannel;

public class WorkTreeWriteFileRequestObserver implements StreamObserver<RpcWriteWorkTreeFileFragment> {

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

    private final WorkTreeManager workTreeManager;
    private final StreamObserver<RpcWriteWorkTreeFileResponse> responseObserver;

    private volatile boolean completed;
    private volatile WritableByteChannel writer;
    private volatile WorkTree workTree;

    public WorkTreeWriteFileRequestObserver(@Nonnull StreamObserver<RpcWriteWorkTreeFileResponse> responseObserver,
                                            @Nonnull WorkTreeManager workTreeManager) {
        this.responseObserver = responseObserver;
        this.workTreeManager = workTreeManager;
    }

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

        } catch (RuntimeException e) {
            if (e instanceof AbstractStatusException) {
                e = ((AbstractStatusException) e).toStatusException();
            }
            sendError(e);
        } catch (IOException e) {
            log.debug("{} Error during work tree completion", workTree, e);
            sendError(new WorkTreeWriteException(e).toStatusException());
        }
        maybeComplete();
    }

    @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", workTree);
        } else {
            log.error("{} Canceling the process; the client has failed", workTree, t);
        }

        // 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 onNext(RpcWriteWorkTreeFileFragment fragment) {
        try {
            switch (fragment.getRequestOneofCase()) {
                case REQUEST:
                    handleRequest(fragment.getRequest());
                    break;
                case CONTENT:
                    handleContent(fragment.getContent());
                    break;
                default:
                    sendError(Status.UNIMPLEMENTED
                            .withDescription("[" + fragment.getRequestOneofCase() + "] is not supported")
                            .asRuntimeException());
            }
        } catch (AbstractStatusException e) {
            sendError(e.toStatusException());
        } catch (RuntimeException | IOException e) {
            log.debug("{} Error during work tree file", workTree, e);
            sendError(new WorkTreeWriteException(e).toStatusException());
        }
    }

    private void handleContent(RpcStreamChunk content) throws IOException {
        if (writer == null) {
            throw Status.INVALID_ARGUMENT
                    .withDescription("A RpcWriteWorkTreeFileRequest must be sent before content")
                    .asRuntimeException();
        }

        for (ByteBuffer buffer : content.getData().asReadOnlyByteBufferList()) {
            writer.write(buffer);
        }
        if (content.getClosed()) {
            writer.close();
            maybeComplete();
        }
    }

    private void handleRequest(RpcWriteWorkTreeFileRequest request) throws IOException {
        if (writer != null) {
            throw Status.INVALID_ARGUMENT
                    .withDescription("Only a single RpcWriteWorkTreeFileRequest may be sent")
                    .asRuntimeException();
        }

        workTree = workTreeManager.getById(request.getWorkTree());
        writer = workTreeManager.writeFile(workTree, request.getPath().toStringUtf8());
    }

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

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