package com.atlassian.bitbucket.mesh.test.context;

import com.atlassian.bitbucket.mesh.client.MeshClient;
import com.atlassian.bitbucket.mesh.client.RepairRepositoryParameters;
import com.atlassian.bitbucket.mesh.layout.StorageLayout;
import com.atlassian.bitbucket.mesh.network.MeshManager;
import com.atlassian.bitbucket.mesh.network.MeshNode;
import com.atlassian.bitbucket.mesh.repair.RepairSummary;
import com.atlassian.bitbucket.mesh.repository.*;
import com.atlassian.bitbucket.mesh.test.annotation.Repo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Field;

/**
 * Specialization of {@link RepoTestExecutionListener} that sync replicas through repair. This is necessary because
 * <ul>
 *     <li>{@link RepoTestExecutionListener} initializes the repository contents by unpacking a zip. This change
 *     will not be visible to remote replicas.</li>
 *     <li>Peer nodes spun up through {@code @MeshCluster} support are reset after each test, which clears all
 *     their on-disk repository data.</li>
 *     <li>The number of peer nodes may have changed since the repository was initialized, and as a result, some
 *     peers may not have the repository yet.</li>
 * </ul>
 */
public class ClusterAwareRepoTestExecutionListener extends RepoTestExecutionListener {

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

    private final MeshClient meshClient;
    private final MeshManager meshManager;

    public ClusterAwareRepoTestExecutionListener(MeshClient meshClient, MeshManager meshManager,
                                                 RepositoryCache repositoryCache,
                                                 RepositoryHashService repositoryHashService,
                                                 RepositoryManager repositoryManager,
                                                 StorageLayout storageLayout) {
        super(repositoryCache, repositoryHashService, repositoryManager, storageLayout);

        this.meshClient = meshClient;
        this.meshManager = meshManager;
    }

    @Override
    protected Object createValue(Field field, Repo annotation) {
        Object result = super.createValue(field, annotation);
        if (result instanceof ManagedRepository) {
            //Make sure that all replicas are in sync with the cached value.
            syncReplicas((ManagedRepository) result);
        }
        return result;
    }

    private void syncReplicas(ManagedRepository repository) {
        meshManager.getNodes().stream()
                .filter(MeshNode::isRemote)
                .forEach(node -> {
                    try {
                        RepairSummary summary = meshClient.repair(node,
                                new RepairRepositoryParameters.Builder(repository.getId())
                                        .timeout(10L)
                                        .build());
                        if (summary.getTargetState() != ReplicaState.CONSISTENT) {
                            log.warn("[{}] Failed to sync replica on {}; state = {}",
                                    repository, node, summary.getTargetState());
                        }
                    } catch (RuntimeException e) {
                        log.warn("[{}] Failed to sync replica on {}", repository, node, e);
                    }
                });
    }
}
