package com.atlassian.bitbucket.mesh.git;

import com.atlassian.bitbucket.mesh.git.commit.Commit;
import com.atlassian.bitbucket.mesh.git.commit.LastModifiedCallback;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcLastModified;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcLastModifiedResponse;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcLastModifiedResponseFragment;
import com.google.protobuf.ByteString;
import io.grpc.stub.StreamObserver;

import javax.annotation.Nonnull;
import java.util.HashMap;
import java.util.Map;

class ObserverLastModifiedCallback
        extends AbstractObserverCallback<RpcLastModifiedResponseFragment>
        implements LastModifiedCallback {

    private final Map<String, Commit> commitsByFile;

    ObserverLastModifiedCallback(StreamObserver<RpcLastModifiedResponseFragment> observer) {
        super(observer);

        commitsByFile = new HashMap<>(FRAGMENT_SIZE, 1.0F);
    }

    @Override
    public void onFileModified(@Nonnull String file, @Nonnull Commit latestCommit) {
        commitsByFile.put(file, latestCommit);
        if (commitsByFile.size() >= FRAGMENT_SIZE) {
            sendFragment();
        }
    }

    @Override
    public void onLatestCommit(@Nonnull Commit latestCommit) {
        //When we get the latest commit, it means all the modified files have already been streamed. If
        //there's a partial fragment buffered, send that first
        sendLastFragment();

        observer.onNext(RpcLastModifiedResponseFragment.newBuilder()
                .setLatestCommit(latestCommit.toCommit())
                .build());
    }

    void onEnd() {
        //If onLatestCommit was called, the last fragment will have already been sent. But if the latest
        //commit couldn't be determined (timeout, some other unexpected condition), there may still be
        //some number of modifications that weren't sent
        sendLastFragment();

        observer.onCompleted();
    }

    private void sendFragment() {
        RpcLastModifiedResponse.Builder builder = RpcLastModifiedResponse.newBuilder();
        commitsByFile.forEach((file, commit) -> builder.addModifications(RpcLastModified.newBuilder()
                .setCommit(commit.toCommit())
                .setPath(ByteString.copyFromUtf8(file))));
        commitsByFile.clear();

        observer.onNext(RpcLastModifiedResponseFragment.newBuilder()
                .setResponse(builder)
                .build());
    }

    private void sendLastFragment() {
        if (!commitsByFile.isEmpty()) {
            sendFragment();
        }
    }
}
