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.push.PushCallback;
import com.atlassian.bitbucket.mesh.git.ref.RefUpdate;
import com.atlassian.bitbucket.mesh.process.ProcessSummary;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcProgressUpdate;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcPushResponseFragment;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcRefUpdate;
import com.google.protobuf.ByteString;
import io.grpc.stub.StreamObserver;

import javax.annotation.Nonnull;
import java.time.Clock;
import java.time.Duration;
import java.time.Instant;

class ObserverPushCallback implements PushCallback {

    private final Clock clock;
    // maximum fragment length in number of characters
    private final long maxFragmentLength;
    private final StreamObserver<RpcPushResponseFragment> observer;
    // minimum time between progress updates in milliseconds
    private final int progressUpdateInterval;

    private Instant lastProgressFlush;
    private RpcPushResponseFragment.Builder progressBuilder = RpcPushResponseFragment.newBuilder();
    private long progressFragmentLength;
    private int refCount;
    private RpcPushResponseFragment.Builder resultBuilder = RpcPushResponseFragment.newBuilder();
    private long resultFragmentLength;

    ObserverPushCallback(long maxFragmentLength, float maxProgressUpdateFrequency,
                         StreamObserver<RpcPushResponseFragment> observer, Clock clock) {
        this.maxFragmentLength = maxFragmentLength;
        this.observer = observer;
        this.clock = clock;

        lastProgressFlush = Instant.MIN;
        progressUpdateInterval = maxProgressUpdateFrequency == 0 ? 0 : Math.round(1000 / maxProgressUpdateFrequency);
    }

    ObserverPushCallback(long maxFragmentLength, float maxProgressUpdateFrequency,
                         StreamObserver<RpcPushResponseFragment> observer) {
        this(maxFragmentLength, maxProgressUpdateFrequency, observer, Clock.systemUTC());
    }

    @Override
    public void onEnd(@Nonnull ProcessSummary summary) {
        if (progressFragmentLength > 0) {
            observer.onNext(progressBuilder.build());
        }
        if (resultFragmentLength > 0) {
            observer.onNext(resultBuilder.build());
        }

        Throwable thrown = summary.getThrown();
        if (refCount == 0 && (thrown != null || summary.getExitCode() != 0)) {
            AbstractStatusException exception = thrown instanceof AbstractStatusException
                    ? (AbstractStatusException) thrown : new CommandFailedException("Push failed", summary);
            observer.onError(exception.toStatusException());
        } else {
            observer.onCompleted();
        }
    }

    @Override
    public void onFailedUpdate(@Nonnull String refId) {
        refCount++;
        resultBuilder.getResultBuilder().addFailedRefIds(ByteString.copyFromUtf8(refId));
        resultFragmentLength += refId.length();
        maybeFlushResult();
    }

    @Override
    public void onProgress(int percentage, @Nonnull String message) {
        RpcProgressUpdate progress = RpcProgressUpdate.newBuilder()
                .setPercentage(percentage)
                .setMessage(message)
                .build();
        progressFragmentLength += progress.getSerializedSize();
        progressBuilder.getProgressUpdateBuilder().addProgressUpdates(progress);
        maybeFlushProgress();
    }

    @Override
    public void onRefUpdate(@Nonnull RefUpdate refUpdate) {
        refCount++;
        RpcRefUpdate result = refUpdate.toRefUpdate();
        resultBuilder.getResultBuilder().addRefUpdates(result);
        resultFragmentLength += result.getSerializedSize();
        maybeFlushResult();
    }

    private void maybeFlushProgress() {
        Instant now = clock.instant();
        if (progressFragmentLength > maxFragmentLength ||
            !Duration.between(lastProgressFlush, now)
                    .minusMillis(progressUpdateInterval)
                    .isNegative()) {
            observer.onNext(progressBuilder.build());
            progressBuilder = RpcPushResponseFragment.newBuilder();
            progressFragmentLength = 0;
            lastProgressFlush = now;
        }
    }

    private void maybeFlushResult() {
        if (resultFragmentLength > maxFragmentLength) {
            observer.onNext(resultBuilder.build());
            resultBuilder = RpcPushResponseFragment.newBuilder();
            resultFragmentLength = 0;
        }
    }
}
