package com.atlassian.bitbucket.mesh.migration;

import com.atlassian.bitbucket.mesh.git.pull.PullRequestRefPredicates;
import com.atlassian.bitbucket.mesh.git.ref.RefUpdate;
import com.atlassian.bitbucket.mesh.repository.BitbucketServerRepository;
import com.atlassian.bitbucket.mesh.repository.ManagedRepository;
import com.atlassian.bitbucket.mesh.repository.Repository;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcProposedRefUpdate;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcTransaction;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcUpdateRefsRequest;
import com.atlassian.bitbucket.mesh.transaction.ReferenceTransaction;
import com.atlassian.bitbucket.mesh.transaction.ReferenceTransactionListener;
import com.atlassian.bitbucket.mesh.util.Person;
import com.google.protobuf.ByteString;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import java.util.Collection;
import java.util.Set;
import java.util.function.BiConsumer;
import java.util.stream.Collectors;

import static com.atlassian.bitbucket.mesh.git.GitConstants.ERR_MIGRATED_TO_MESH;
import static com.atlassian.bitbucket.mesh.util.MoreCollectors.toImmutableSet;
import static java.util.Objects.requireNonNull;

/**
 * {@link ReferenceTransactionListener} that records reference transactions in the hierarchy transaction log if
 * that log file exists on disk.
 *
 * This rejects reference transactions that occur on a repository whose hierarchy has already been migrated to Mesh. This
 * could be because the transaction started before/during the migration of the repository and finished after it has been
 * migrated to a Mesh node. If we allowed the transaction to finish, then the ref update wouldn't be propagated to its target
 * and therefore, lost.
 */
public class SidecarMigrationTransactionListener implements ReferenceTransactionListener {

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

    private final MigrationStateManager migrationStateManager;

    public SidecarMigrationTransactionListener(MigrationStateManager migrationStateManager) {
        this.migrationStateManager = migrationStateManager;
    }

    @Override
    public void onReferenceTransaction(@Nonnull ReferenceTransaction transaction) {
        ifMigrationStatePresent(transaction.getRepository(), (repository, state) -> {
            switch (transaction.getState()) {
                case PREPARED:
                    if (state.isCompleted()) {
                        //The repository has already been migrated to a remote partition
                        transaction.reject(ERR_MIGRATED_TO_MESH);
                    }
                    break;
                case COMMITTED:
                    maybeLogTransaction(repository, state, transaction);
                    break;
            }
        });
    }

    public void ifMigrationStatePresent(Repository repository,
                                        BiConsumer<BitbucketServerRepository, HierarchyMigrationState> callback) {
        if (repository instanceof BitbucketServerRepository) {
            BitbucketServerRepository bitbucketServerRepository = (BitbucketServerRepository) repository;
            migrationStateManager.getForHierarchy(bitbucketServerRepository.getHierarchyId())
                    .ifPresent(state -> callback.accept(bitbucketServerRepository, state));
        }
    }

    private static void maybeLogTransaction(BitbucketServerRepository repository, HierarchyMigrationState state,
                                            ReferenceTransaction transaction) {
        if (!state.isRunning()) {
            return;
        }

        String hierarchyId = repository.getHierarchyId();
        Set<RpcProposedRefUpdate> refUpdates = toProposedRefUpdates(transaction.getRefUpdates());
        if (refUpdates.isEmpty()) {
            //Only updates to PR refs
            return;
        }

        try {
            String remoteId = ManagedRepository.formatId(
                    state.getTargetPartition(), hierarchyId, repository.getBitbucketId());
            state.getTransactionLog().append(RpcTransaction.newBuilder()
                    .setPerson(Person.toPerson(transaction.getUser()))
                    .setUpdateRefs(RpcUpdateRefsRequest.newBuilder()
                            .addAllRefUpdates(refUpdates)
                            .setRepository(remoteId)
                            .setTransactional(true))
                    .setRepository(remoteId)
                    .build());
        } catch (RuntimeException e) {
            log.error("[{}] Failed to record ref-updates to ({}) in the migration transaction log." +
                            "The change may be lost after migration completes!",
                    repository, transaction.getRefUpdates().stream()
                            .map(RefUpdate::getRefId)
                            .collect(Collectors.joining(","))
            );
        }
    }

    @Nonnull
    public static Set<RpcProposedRefUpdate> toProposedRefUpdates(@Nonnull Collection<RefUpdate> refUpdates) {
        requireNonNull(refUpdates, "refUpdates");

        //RpcTransactions are recorded from the beginning of the migration of the hierarchy. This means that some
        //ref changes may end be applied to the repository _before_ its refs are uploaded to Mesh. As a result, when
        //the transaction is replayed, the ref may not be in the expected state (mismatch in the oldHash).
        //We _intentionally_ do not set the oldHash here, which ensures that when the transaction is replayed, the
        //ref is updated regardless of the current value of oldHash.
        return refUpdates.stream()
                //Pull request ref updates are handled through separate transaction types
                .filter(PullRequestRefPredicates::isNotPullRequestRefUpdate)
                .map(refUpdate -> RpcProposedRefUpdate.newBuilder()
                        .setRefId(ByteString.copyFromUtf8(refUpdate.getRefId()))
                        .setNewHash(refUpdate.getNewHash())
                        .build())
                .collect(toImmutableSet());
    }
}
