package com.atlassian.bitbucket.mesh.git;

import com.atlassian.bitbucket.mesh.AbstractStatusException;
import com.atlassian.bitbucket.mesh.git.exception.CommandFailedException;
import com.atlassian.bitbucket.mesh.git.fetch.FetchCallback;
import com.atlassian.bitbucket.mesh.git.ref.RefUpdate;
import com.atlassian.bitbucket.mesh.process.ProcessSummary;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcFetchResponseFragment;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcRefUpdate;
import com.google.protobuf.ByteString;
import io.grpc.stub.StreamObserver;

import javax.annotation.Nonnull;

class ObserverFetchCallback implements FetchCallback {

    // maximum fragment length in number of characters
    private final long maxFragmentLength;
    private final StreamObserver<RpcFetchResponseFragment> observer;

    private RpcFetchResponseFragment.Builder builder = RpcFetchResponseFragment.newBuilder();
    private long fragmentLength;
    private int refCount;

    ObserverFetchCallback(long maxFragmentLength, StreamObserver<RpcFetchResponseFragment> observer) {
        this.maxFragmentLength = maxFragmentLength;
        this.observer = observer;
    }

    @Override
    public void onEnd(@Nonnull ProcessSummary summary) {
        boolean hasFragments = fragmentLength > 0;
        if (hasFragments) {
            callOnNext();
        }
        Throwable thrown = summary.getThrown();
        if (refCount == 0 && (thrown != null || summary.getExitCode() != 0)) {
            AbstractStatusException exception = thrown instanceof AbstractStatusException
                    ? (AbstractStatusException) thrown : new CommandFailedException("Fetch failed", summary);
            observer.onError(exception.toStatusException());
        } else {
            observer.onCompleted();
        }
    }

    @Override
    public void onFailedUpdate(@Nonnull String refId) {
        refCount++;
        builder.addFailedRefIds(ByteString.copyFromUtf8(refId));
        fragmentLength += refId.length();
        maybeCallOnNext();
    }

    @Override
    public void onRefUpdate(@Nonnull RefUpdate refUpdate) {
        refCount++;
        RpcRefUpdate result = refUpdate.toRefUpdate();
        builder.addRefUpdates(result);
        fragmentLength += result.getSerializedSize();
        maybeCallOnNext();
    }

    private void callOnNext() {
        observer.onNext(builder.build());
        builder = RpcFetchResponseFragment.newBuilder();
        fragmentLength = 0;
    }

    private void maybeCallOnNext() {
        if (fragmentLength > maxFragmentLength) {
            callOnNext();
        }
    }
}
