package com.atlassian.bitbucket.mesh.git;

import com.atlassian.bitbucket.mesh.AbstractStatusException;
import com.atlassian.bitbucket.mesh.event.EventPublisher;
import com.atlassian.bitbucket.mesh.execution.ExecutionManager;
import com.atlassian.bitbucket.mesh.git.builder.GitProcessBuilderFactory;
import com.atlassian.bitbucket.mesh.git.builder.indexpack.IndexPackStdioHandler;
import com.atlassian.bitbucket.mesh.git.exception.InvalidArgumentException;
import com.atlassian.bitbucket.mesh.process.ProcessSummary;
import com.atlassian.bitbucket.mesh.repository.ManagedRepository;
import com.atlassian.bitbucket.mesh.repository.ObjectsUploadedToSharedObjectsEvent;
import com.atlassian.bitbucket.mesh.repository.Repository;
import com.atlassian.bitbucket.mesh.repository.RepositoryManager;
import com.atlassian.bitbucket.mesh.rpc.ZeroCopyUtils;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcStreamChunk;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcUploadPackFileFragment;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcUploadPackFileRequest;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcUploadPackFileResponse;
import io.grpc.Status;
import io.grpc.stub.StreamObserver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import java.nio.file.Path;
import java.util.concurrent.Future;

import static com.atlassian.bitbucket.mesh.git.GitConstants.PATH_OBJECTS;

class UploadPackFileRequestObserver implements StreamObserver<RpcUploadPackFileFragment> {

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

    private final GitAgent agent;
    private final GitProcessBuilderFactory builderFactory;
    private final EventPublisher eventPublisher;
    private final ExecutionManager executionManager;
    private final RepositoryManager repositoryManager;
    private final StreamObserver<RpcUploadPackFileResponse> responseObserver;

    private volatile Future<Path> indexPackFuture;
    private volatile Runnable indexPackListener;
    private volatile IndexPackStdioHandler indexPackHandler;
    private volatile Repository repository;
    private volatile RpcUploadPackFileRequest request;

    UploadPackFileRequestObserver(GitAgent agent, GitProcessBuilderFactory builderFactory,
                                  EventPublisher eventPublisher, ExecutionManager executionManager,
                                  RepositoryManager repositoryManager,
                                  StreamObserver<RpcUploadPackFileResponse> responseObserver) {
        this.agent = agent;
        this.builderFactory = builderFactory;
        this.eventPublisher = eventPublisher;
        this.executionManager = executionManager;
        this.repositoryManager = repositoryManager;
        this.responseObserver = responseObserver;
    }

    @Override
    public void onCompleted() {
        if (indexPackHandler != null) {
            //Close stdin in case the closed flag was not set on the last chunk, or no chunks were sent.
            //This is a no-op if stdin has already been closed.
            indexPackHandler.closeStdin();
        }
    }

    @Override
    public void onError(Throwable t) {
        if (indexPackFuture != null) {
            if (indexPackFuture.isDone()) {
                log.debug("[{}] Error processing UploadPackFile, but index-pack has already finished", repository, t);
                return;
            }

            indexPackFuture.cancel(true);

            Status status = Status.fromThrowable(t);
            if (status.getCode() == Status.Code.CANCELLED) {
                log.debug("[{}] Client canceled UploadPackFile", repository);
            } else {
                log.warn("[{}] Error processing UploadPackFile", repository, t);
            }
        } else {
            log.warn("Error processing UploadPackFile", t);
        }
    }

    @Override
    public void onNext(RpcUploadPackFileFragment fragment) {
        switch (fragment.getRequestOneofCase()) {
            case CHUNK:
                processChunk(fragment.getChunk());
                break;
            case REQUEST:
                processRequest(fragment.getRequest());
                break;
            default:
                throw new InvalidArgumentException("[" + fragment.getRequestOneofCase() + "] is not supported");
        }
    }

    private void processChunk(RpcStreamChunk chunk) {
        if (request == null) {
            throw new InvalidArgumentException("An RpcUploadPackFileRequest must be sent before content");
        }

        if (indexPackHandler == null) {
            Repository destination = repository;
            if (repository instanceof ManagedRepository && !agent.hasObjects(repository)) {
                //For empty managed repositories, write the pack immediately into the shared-objects repository.
                //As long as the managed repository is empty, this is safe because the pack we're about to process
                //can only depend on objects present in the shared-objects repository.
                destination = ((ManagedRepository) repository).getSharedObjects();
                //After index-pack completes, we raise an event to ensure the objects are protected from GC
                indexPackListener = () -> eventPublisher.publish(
                        new ObjectsUploadedToSharedObjectsEvent((ManagedRepository) repository));
            }
            indexPackHandler = new IndexPackStdioHandler(destination.getPath().resolve(PATH_OBJECTS)) {

                @Override
                public void onComplete(@Nonnull ProcessSummary summary) {
                    try {
                        super.onComplete(summary);
                        responseObserver.onNext(RpcUploadPackFileResponse.getDefaultInstance());
                        responseObserver.onCompleted();
                        if (indexPackListener != null) {
                            indexPackListener.run();
                        }
                    } catch (RuntimeException e) {
                        log.warn("[{}] git index-pack failed (exit code = {}): {}",
                                repository, summary.getExitCode(), e.getMessage());
                        if (e instanceof AbstractStatusException) {
                            responseObserver.onError(((AbstractStatusException) e).toStatusException());
                        } else {
                            responseObserver.onError(e);
                        }
                    }
                }
            };
            indexPackFuture = builderFactory.builder(destination).indexPack()
                    .fixThin(true)
                    .stdin(true)
                    .strict(true)
                    .withTimeouts(request.hasTimeouts() ? request.getTimeouts() : null)
                    .build(indexPackHandler)
                    .start();

            executionManager.getContextOrThrow().addCleanupCallback(() -> {
                if (!indexPackFuture.isDone()) {
                    indexPackFuture.cancel(true);
                }
            });
        }
        indexPackHandler.writeStdin(chunk.getData(), chunk.getClosed());
    }

    private void processRequest(RpcUploadPackFileRequest request) {
        if (this.request != null) {
            throw new InvalidArgumentException("Only one RpcUploadPackFileRequest is supported");
        }
        //As a result of zero-copy optimizations, any ByteStrings in request are backed by the underlying Netty
        //buffer, which will get recycled after this method returns. While RpcUploadPackFileRequest does not currently
        //have any bytes fields on it, we'll create a deep copy to future-proof it.
        this.request = ZeroCopyUtils.deepCopy(request);

        repository = repositoryManager.getById(request.getRepository());
    }
}
