package com.atlassian.bitbucket.mesh.git;

import com.atlassian.bitbucket.mesh.config.ApplicationConfig;
import com.atlassian.bitbucket.mesh.config.GrpcConfig;
import com.atlassian.bitbucket.mesh.coordination.LogicalClock;
import com.atlassian.bitbucket.mesh.credentials.Credentials;
import com.atlassian.bitbucket.mesh.credentials.CredentialsHelper;
import com.atlassian.bitbucket.mesh.event.EventPublisher;
import com.atlassian.bitbucket.mesh.execution.GrpcAsyncExecutor;
import com.atlassian.bitbucket.mesh.git.builder.GitProcessBuilderFactory;
import com.atlassian.bitbucket.mesh.git.builder.fetch.GitFetchTagMode;
import com.atlassian.bitbucket.mesh.git.exception.InvalidArgumentException;
import com.atlassian.bitbucket.mesh.git.fetch.FetchManager;
import com.atlassian.bitbucket.mesh.git.fetch.FetchParameters;
import com.atlassian.bitbucket.mesh.git.gc.GarbageCollectionManager;
import com.atlassian.bitbucket.mesh.git.gc.GarbageCollectionParameters;
import com.atlassian.bitbucket.mesh.git.porcelain.CountObjectsParameters;
import com.atlassian.bitbucket.mesh.git.porcelain.GitPorcelain;
import com.atlassian.bitbucket.mesh.git.pull.PullRequestManager;
import com.atlassian.bitbucket.mesh.git.push.PushManager;
import com.atlassian.bitbucket.mesh.git.push.PushParameters;
import com.atlassian.bitbucket.mesh.git.transcode.TranscodeService;
import com.atlassian.bitbucket.mesh.grpc.BackoffStreamObserver;
import com.atlassian.bitbucket.mesh.repair.InteractiveRepairHelper;
import com.atlassian.bitbucket.mesh.repair.RepairSource;
import com.atlassian.bitbucket.mesh.repair.RepairTarget;
import com.atlassian.bitbucket.mesh.repository.*;
import com.atlassian.bitbucket.mesh.rpc.v1.*;
import com.atlassian.bitbucket.mesh.transaction.RepositoryConsistencySummary;
import com.atlassian.bitbucket.mesh.transaction.TransactionManager;
import com.google.protobuf.ByteString;
import io.grpc.stub.StreamObserver;

import java.time.Duration;
import java.util.Map;
import java.util.concurrent.ExecutorService;

import static com.atlassian.bitbucket.mesh.rpc.util.ByteStringUtils.emptyToNull;
import static com.atlassian.bitbucket.mesh.rpc.util.ByteStringUtils.toStringsLazily;

public class GrpcRepositoryService extends RepositoryServiceGrpc.RepositoryServiceImplBase {

    private final GitAgent agent;
    private final GitProcessBuilderFactory builderFactory;
    private final ApplicationConfig config;
    private final CredentialsHelper credentialsHelper;
    private final EventPublisher eventPublisher;
    private final GrpcAsyncExecutor executor;
    private final ExecutorService executorService;
    private final FetchManager fetchManager;
    private final GarbageCollectionManager garbageCollectionManager;
    private final LogicalClock logicalClock;
    private final GitPorcelain porcelain;
    private final PullRequestManager pullRequestManager;
    private final PushManager pushManager;
    private final InteractiveRepairHelper repairHelper;
    private final ReplicaStateRegistry replicaStateRegistry;
    private final RepositoryHashService repositoryHashService;
    private final RepositoryManager repositoryManager;
    private final TransactionManager transactionManager;
    private final TranscodeService transcodeService;

    public GrpcRepositoryService(GitAgent agent, GitProcessBuilderFactory builderFactory, ApplicationConfig config,
                                 CredentialsHelper credentialsHelper, EventPublisher eventPublisher,
                                 GrpcAsyncExecutor executor, ExecutorService executorService, FetchManager fetchManager,
                                 GarbageCollectionManager garbageCollectionManager, LogicalClock logicalClock,
                                 GitPorcelain porcelain, PullRequestManager pullRequestManager, PushManager pushManager,
                                 InteractiveRepairHelper repairHelper, ReplicaStateRegistry replicaStateRegistry,
                                 RepositoryHashService repositoryHashService, RepositoryManager repositoryManager,
                                 TransactionManager transactionManager, TranscodeService transcodeService) {
        this.agent = agent;
        this.builderFactory = builderFactory;
        this.config = config;
        this.credentialsHelper = credentialsHelper;
        this.eventPublisher = eventPublisher;
        this.executor = executor;
        this.executorService = executorService;
        this.fetchManager = fetchManager;
        this.garbageCollectionManager = garbageCollectionManager;
        this.logicalClock = logicalClock;
        this.porcelain = porcelain;
        this.pushManager = pushManager;
        this.pullRequestManager = pullRequestManager;
        this.repairHelper = repairHelper;
        this.replicaStateRegistry = replicaStateRegistry;
        this.repositoryHashService = repositoryHashService;
        this.repositoryManager = repositoryManager;
        this.transactionManager = transactionManager;
        this.transcodeService = transcodeService;
    }

    @Override
    public void bulkVerifyRepositoryIntegrity(RpcBulkVerifyRepositoryIntegrityRequest request,
                                              StreamObserver<RpcBulkVerifyRepositoryIntegrityResponse> responseObserver) {
        RpcBulkVerifyRepositoryIntegrityResponse.Builder builder = RpcBulkVerifyRepositoryIntegrityResponse.newBuilder();
        request.getRepositoryRequestsList().forEach(repoRequest -> {
            String repositoryId = repoRequest.getRepository();
            Repository repository = repositoryManager.findById(repositoryId);
            if (repository != null) {
                pullRequestManager.checkIntegrity(repository, repoRequest.getLatestPullRequestId())
                        .ifPresent(builder::addIntegrityIssues);
            }
        });

        responseObserver.onNext(builder.build());
        responseObserver.onCompleted();
    }

    @Override
    public void create(RpcCreateRepositoryRequest request, StreamObserver<RpcCreateRepositoryResponse> observer) {
        CreateRepositoryParameters.Builder builder = new CreateRepositoryParameters.Builder(request.getRepository())
                .defaultBranch(emptyToNull(request.getDefaultBranch()))
                .metadata(request.getMetadataMap());
        if (request.hasOrigin()) {
            builder.origin(repositoryManager.getById(request.getOrigin()));
        }
        if (request.hasTimeouts()) {
            builder.timeouts(request.getTimeouts());
        }

        repositoryManager.create(builder.build());
        observer.onNext(RpcCreateRepositoryResponse.getDefaultInstance());
        observer.onCompleted();
    }

    @Override
    public void delete(RpcDeleteRepositoryRequest request, StreamObserver<RpcDeleteRepositoryResponse> observer) {
        Repository repository = repositoryManager.getById(request.getRepository());

        DeleteRepositoryParameters.Builder builder = new DeleteRepositoryParameters.Builder(repository);
        if (request.hasHierarchyMembership()) {
            RpcHierarchyMembership membership = request.getHierarchyMembership();
            builder.onlyMember(membership.getOnlyMember());
            if (membership.hasOrigin()) {
                builder.origin(repositoryManager.getById(membership.getOrigin()));
                if (membership.hasOnlyForkOfOrigin()) {
                    builder.onlyForkOfOrigin(membership.getOnlyForkOfOrigin());
                }
            }
            membership.getForksList().forEach(forkId -> builder.forks(repositoryManager.getById(forkId)));
        }

        repositoryManager.delete(builder.build());
        observer.onNext(RpcDeleteRepositoryResponse.getDefaultInstance());
        observer.onCompleted();
    }

    @Override
    public void fetch(RpcFetchRequest request, StreamObserver<RpcFetchResponseFragment> responseObserver) {
        Repository repository = repositoryManager.getById(request.getRepository());

        Credentials credentials = null;
        RpcAnyRepository source = request.getSourceRepository();
        String sourceUrl;
        switch (source.getRepositoryOneofCase()) {
            case REPOSITORY:
                // it's a local repository
                Repository sourceRepository = repositoryManager.getById(source.getRepository());
                sourceUrl = sourceRepository.getRelativePath(repository).toString();
                break;
            case REMOTE_REPOSITORY:
                RpcRemoteRepository remoteRepository = source.getRemoteRepository();
                sourceUrl = remoteRepository.getUrl();
                credentials = credentialsHelper.getCredentials(remoteRepository);
                break;
            case REPOSITORYONEOF_NOT_SET:
                throw new InvalidArgumentException("No repository was provided");
            default:
                throw new InvalidArgumentException("AnyRepository of type " + source.getRepositoryOneofCase() +
                        " is not supported");
        }

        FetchParameters.Builder builder = new FetchParameters.Builder(repository, request.getTimeouts())
                .atomic(request.getAtomic())
                .credentials(credentials)
                .prune(request.getPrune())
                .refspecs(toStringsLazily(request.getRefspecsList()))
                .sourceUrl(sourceUrl)
                .tagMode(request.getNoTags() ? GitFetchTagMode.NO_TAGS : GitFetchTagMode.DEFAULT);
        if (request.hasOptions()) {
            builder.withEnvironment(toStringsLazily(request.getOptions().getEnvironmentMap()));
        }
        FetchParameters parameters = builder.build();

        ObserverFetchCallback callback = new ObserverFetchCallback(config.getGrpc().getFragmentSize(),
                new BackoffStreamObserver<>(responseObserver));

        executor.execute(responseObserver, () -> fetchManager.fetch(parameters, callback));
    }

    @Override
    public void getDefaultBranch(RpcGetDefaultBranchRequest request,
                                 StreamObserver<RpcGetDefaultBranchResponse> responseObserver) {
        Repository repository = repositoryManager.getById(request.getRepository());
        String head = agent.getHead(repository);

        responseObserver.onNext(RpcGetDefaultBranchResponse.newBuilder()
                .setBranchId(ByteString.copyFromUtf8(head))
                .build());
        responseObserver.onCompleted();
    }

    @Override
    public void getDetails(RpcGetRepositoryDetailsRequest request,
                            StreamObserver<RpcGetRepositoryDetailsResponse> responseObserver) {
        Repository repository = repositoryManager.getById(request.getRepository());

        String defaultBranch = repositoryManager.getDefaultBranch(repository);
        RpcGetRepositoryDetailsResponse.Builder builder = RpcGetRepositoryDetailsResponse.newBuilder()
                .putAllMetadata(repositoryManager.getMetadata(repository))
                .setTranscodeEnabled(transcodeService.isEnabled(repository));
        if (defaultBranch != null) {
            builder.setDefaultBranch(ByteString.copyFromUtf8(defaultBranch));
        }
        responseObserver.onNext(builder.build());
        responseObserver.onCompleted();
    }

    @Override
    public void getReplicaDetails(RpcGetRepositoryReplicaDetailsRequest request,
                                  StreamObserver<RpcGetRepositoryReplicaDetailsResponse> responseObserver) {
        responseObserver.onNext(RpcGetRepositoryReplicaDetailsResponse.newBuilder()
                .setDetails(repositoryManager.getReplicaDetails(request.getRepository()).toRpc())
                .build());
        responseObserver.onCompleted();
    }

    @Override
    public void getRepositories(RpcGetRepositoriesRequest request,
                                StreamObserver<RpcGetRepositoriesResponseFragment> responseObserver) {
        long maxFragmentSize = config.getGrpc().getMaxFragmentSize();
        try (GetRepositoriesHandler handler = new GetRepositoriesHandler(responseObserver, maxFragmentSize)) {
            Map<String, Long> versionById = repositoryManager.getVersions(new GetVersionsParameters.Builder()
                    .partitions(request.getPartitionsList())
                    .build());
            versionById.entrySet().stream()
                    .skip(request.getSkip())
                    .limit(request.getLimit() > 0 ? request.getLimit() : Long.MAX_VALUE)
                    .forEach(entry -> handler.onRepository(entry.getKey(), entry.getValue()));
        }
    }

    @Override
    public void getSize(RpcGetRepositorySizeRequest request,
                        StreamObserver<RpcGetRepositorySizeResponse> responseObserver) {
        Repository repository = repositoryManager.getById(request.getRepository());
        RpcGetRepositorySizeResponse.Builder responseBuilder = RpcGetRepositorySizeResponse.newBuilder();

        RepositorySize repositorySize = porcelain.countObjects(new CountObjectsParameters.Builder(repository)
                .timeouts(request.getTimeouts())
                .build());
        responseBuilder.setSize(repositorySize.toRepositorySize());

        if (repository instanceof ManagedRepository) {
            SharedObjectsRepository sharedObjectsRepo = ((ManagedRepository) repository).getSharedObjects();
            RepositorySize sharedSize = porcelain.countObjects(new CountObjectsParameters.Builder(sharedObjectsRepo)
                    .timeouts(request.getTimeouts())
                    .build());
            responseBuilder.setSharedSize(sharedSize.toRepositorySize());
        }

        responseObserver.onNext(responseBuilder.build());
        responseObserver.onCompleted();
    }

    @Override
    public void isEmpty(RpcIsRepositoryEmptyRequest request,
                        StreamObserver<RpcIsRepositoryEmptyResponse> responseObserver) {
        Repository repository = repositoryManager.getById(request.getRepository());
        boolean empty = agent.isEmpty(repository);

        responseObserver.onNext(RpcIsRepositoryEmptyResponse.newBuilder()
                .setEmpty(empty)
                .build());
        responseObserver.onCompleted();
    }

    @Override
    public void isTranscodeEnabled(RpcIsTranscodeEnabledRequest request,
                                   StreamObserver<RpcIsTranscodeEnabledResponse> responseObserver) {
        Repository repository = repositoryManager.getById(request.getRepository());
        boolean enabled = transcodeService.isEnabled(repository);

        responseObserver.onNext(RpcIsTranscodeEnabledResponse.newBuilder()
                .setEnabled(enabled)
                .build());
        responseObserver.onCompleted();
    }

    @Override
    public void push(RpcPushRequest request, StreamObserver<RpcPushResponseFragment> responseObserver) {
        Repository repository = repositoryManager.getById(request.getRepository());

        RpcRemoteRepository remoteRepository = request.getTargetRepository();
        String targetUrl = remoteRepository.getUrl();
        Credentials credentials = credentialsHelper.getCredentials(remoteRepository);

        PushParameters.Builder builder = new PushParameters.Builder(repository, request.getTimeouts())
                .credentials(credentials)
                .force(request.getForce())
                .prune(request.getPrune())
                .refspecs(toStringsLazily(request.getRefspecsList()))
                .targetUrl(targetUrl);
        if (request.hasOptions()) {
            builder.withEnvironment(toStringsLazily(request.getOptions().getEnvironmentMap()));
        }
        PushParameters parameters = builder.build();

        GrpcConfig config = this.config.getGrpc();
        ObserverPushCallback callback = new ObserverPushCallback(config.getFragmentSize(),
                config.getMaxProgressUpdateFrequency(), new BackoffStreamObserver<>(responseObserver));
        executor.execute(responseObserver, () -> pushManager.push(parameters, callback));
    }

    @Override
    public StreamObserver<RpcRepairSourceFragment> repair(StreamObserver<RpcRepairTargetFragment> repairSource) {
        return new RepairTarget(eventPublisher, config.getGrpc(), logicalClock, repairHelper,
                new BackoffStreamObserver<>(repairSource), config.getRepair().getTimeout());
    }

    @Override
    public StreamObserver<RpcRepairTargetFragment> repairMe(StreamObserver<RpcRepairSourceFragment> repairTarget) {
        return new RepairSource(builderFactory, executorService, garbageCollectionManager, config.getGrpc(),
                logicalClock, porcelain, new BackoffStreamObserver<>(repairTarget), replicaStateRegistry,
                repositoryHashService, repositoryManager, transcodeService, config.getRepair().getTimeout());
    }

    @Override
    public void scheduleGarbageCollection(RpcScheduleGarbageCollectionRequest request,
                                          StreamObserver<RpcScheduleGarbageCollectionResponse> responseObserver) {
        responseObserver.onNext(RpcScheduleGarbageCollectionResponse.getDefaultInstance());
        responseObserver.onCompleted();

        garbageCollectionManager.collectGarbage(buildGarbageCollectionParameters(request));
    }

    @Override
    public void setDefaultBranch(RpcSetDefaultBranchRequest request,
                                 StreamObserver<RpcSetDefaultBranchResponse> responseObserver) {
        Repository repository = repositoryManager.getById(request.getRepository());
        repositoryManager.setDefaultBranch(repository, request.getBranchId().toStringUtf8());

        responseObserver.onNext(RpcSetDefaultBranchResponse.getDefaultInstance());
        responseObserver.onCompleted();
    }

    @Override
    public void setGarbageCollectionPaused(RpcSetGarbageCollectionPausedRequest request,
                                           StreamObserver<RpcSetGarbageCollectionPausedResponse> responseObserver) {
        Repository repository = repositoryManager.getById(request.getRepository());
        boolean result;
        if (request.getPaused()) {
            result = garbageCollectionManager.pauseIndefinitely(repository, Duration.ofSeconds(request.getTimeout()));
        } else {
            result = garbageCollectionManager.resume(repository);
        }

        responseObserver.onNext(RpcSetGarbageCollectionPausedResponse.newBuilder().setResult(result).build());
        responseObserver.onCompleted();
    }

    @Override
    public void setMetadata(RpcSetMetadataRequest request, StreamObserver<RpcSetMetadataResponse> responseObserver) {
        Repository repository = repositoryManager.getById(request.getRepository());
        repositoryManager.setMetadata(repository, request.getMetadataMap());

        responseObserver.onNext(RpcSetMetadataResponse.getDefaultInstance());
        responseObserver.onCompleted();
    }

    @Override
    public void setTranscodeEnabled(RpcSetTranscodeEnabledRequest request,
                                    StreamObserver<RpcSetTranscodeEnabledResponse> responseObserver) {
        Repository repository = repositoryManager.getById(request.getRepository());
        transcodeService.setEnabled(repository, request.getEnabled());

        responseObserver.onNext(RpcSetTranscodeEnabledResponse.getDefaultInstance());
        responseObserver.onCompleted();
    }

    @Override
    public void verifyConsistency(RpcVerifyRepositoryConsistencyRequest request,
                                  StreamObserver<RpcVerifyRepositoryConsistencyResponse> responseObserver) {
        RepositoryConsistencySummary summary = transactionManager.verifyConsistency(request.getRepository());
        if (summary == null) {
            //The repository isn't managed
            responseObserver.onNext(RpcVerifyRepositoryConsistencyResponse.getDefaultInstance());
        } else {
            responseObserver.onNext(RpcVerifyRepositoryConsistencyResponse.newBuilder()
                    .setSummary(summary.toRpc())
                    .build());
        }
        responseObserver.onCompleted();
    }

    private GarbageCollectionParameters buildGarbageCollectionParameters(RpcScheduleGarbageCollectionRequest request) {
        Repository repository = repositoryManager.getById(request.getRepository());
        return new GarbageCollectionParameters.Builder(repository)
                .added(request.getAdded())
                .deleted(request.getDeleted())
                .staleOnly(request.getStaleOnly())
                .updated(request.getUpdated())
                .build();
    }

    private static class GetRepositoriesHandler implements AutoCloseable {

        private final RpcGetRepositoriesResponseFragment.Builder builder;
        private final long maxFragmentSize;
        private final StreamObserver<RpcGetRepositoriesResponseFragment> responseObserver;
        private long size;

        GetRepositoriesHandler(StreamObserver<RpcGetRepositoriesResponseFragment> responseObserver,
                               long maxFragmentSize) {
            this.maxFragmentSize = maxFragmentSize;
            this.responseObserver = responseObserver;

            builder = RpcGetRepositoriesResponseFragment.newBuilder();
        }

        public void onRepository(String id, long version) {
            RpcRepository rpc = RpcRepository.newBuilder()
                    .setId(id)
                    .setVersion(version)
                    .build();
            if (size + rpc.getSerializedSize() > maxFragmentSize) {
                emit();
            }
            builder.addRepositories(rpc);
            size += rpc.getSerializedSize();
        }

        @Override
        public void close() {
            if (size > 0L) {
                emit();
            }
            responseObserver.onCompleted();
        }

        private void emit() {
            responseObserver.onNext(builder.build());
            //Reset the builder
            builder.clearRepositories();
            size = 0L;
        }
    }
}
