package com.atlassian.bitbucket.mesh.repository;

import com.atlassian.bitbucket.mesh.coordination.LogicalClock;
import com.atlassian.bitbucket.mesh.event.EventPublisher;
import com.atlassian.bitbucket.mesh.management.SidebandManager;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcRepositoryReplicaObservation;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcSidebandFollowerFragment;
import com.google.common.collect.ImmutableMap;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.*;
import org.mockito.junit.MockitoJUnitRunner;

import static com.atlassian.bitbucket.mesh.repository.ReplicaState.*;
import static org.junit.Assert.*;
import static org.mockito.Mockito.*;

@RunWith(MockitoJUnitRunner.class)
public class DefaultReplicaStateRegistryTest {

    private static final long NOW = 67944123L;
    private static final String REPOSITORY_ID = "p/0013/h/abcdef52123791279129/r/79234";
    private static final long VERSION = 2387L;

    @Captor
    private ArgumentCaptor<ReplicaStateUpdatedEvent> eventCaptor;
    @Mock
    private EventPublisher eventPublisher;
    @Captor
    private ArgumentCaptor<RpcSidebandFollowerFragment> fragmentCaptor;
    @Mock
    private LogicalClock logicalClock;
    @InjectMocks
    private DefaultReplicaStateRegistry registry;
    @Mock
    private ManagedRepository repository;
    @Mock
    private SidebandManager sidebandManager;
    private long timestamp;

    @Before
    public void setUp() {
        when(repository.accept(any())).thenAnswer(invocation -> {
            RepositoryVisitor<?> visitor = invocation.getArgument(0);
            return visitor.visit(repository);
        });
        when(repository.getId()).thenReturn(REPOSITORY_ID);
        when(repository.getVersion()).thenReturn(VERSION);
        when(logicalClock.tick()).thenAnswer(ignored -> ++timestamp);
        lenient().when(logicalClock.get()).thenAnswer(ignored -> timestamp);
        timestamp = NOW;
    }

    @Test
    public void testCanonicalIdUsedInternally() {
        when(repository.getId()).thenReturn(REPOSITORY_ID + "?q=quarantine-1");

        registry.onReplicaStateObserved(new ReplicaStateObservedEvent(REPOSITORY_ID + "?q=quarantine-2",
                new ReplicaStateObservation(UNKNOWN, NOW, VERSION)));
        assertEquals(UNKNOWN, registry.getState(repository));
        verifyStateUpdated(UNKNOWN, VERSION);
    }

    @Test
    public void testConsistentObservationIgnoredIfAlreadyConsistent() {
        assertEquals(CONSISTENT, registry.getState(repository));
        registry.onReplicaStateObserved(
                new ReplicaStateObservedEvent(REPOSITORY_ID, new ReplicaStateObservation(CONSISTENT, NOW, VERSION)));
        assertEquals(CONSISTENT, registry.getState(repository));
        verifyNoInteractions(eventPublisher);
    }

    @Test
    public void testConsistentObservationTriggersStateChangeIfInconsistent() {
        //Mark the repository as inconsistent (60 ticks ago)
        registry.onReplicaStateObserved(new ReplicaStateObservedEvent(REPOSITORY_ID,
                new ReplicaStateObservation(INCONSISTENT, NOW - 60L, VERSION)));
        assertEquals(INCONSISTENT, registry.getState(repository));
        verifyStateUpdated(INCONSISTENT, VERSION);

        //Trigger a CONSISTENT observation
        registry.onReplicaStateObserved(
                new ReplicaStateObservedEvent(REPOSITORY_ID, new ReplicaStateObservation(CONSISTENT, NOW, VERSION)));
        assertEquals(CONSISTENT, registry.getState(repository));
        verifyStateUpdated(CONSISTENT, VERSION);
    }

    @Test
    public void testCount() {
        assertEquals(0, registry.countByState(INCONSISTENT));

        registry.onReplicaStateObserved(new ReplicaStateObservedEvent(REPOSITORY_ID,
                new ReplicaStateObservation(INCONSISTENT, logicalClock.tick(), VERSION)));
        assertEquals(1, registry.countByState(INCONSISTENT));

        registry.onReplicaStateObserved(new ReplicaStateObservedEvent("repo-2",
                new ReplicaStateObservation(INCONSISTENT, logicalClock.tick(), VERSION)));
        assertEquals(2, registry.countByState(INCONSISTENT));

        registry.onReplicaStateObserved(new ReplicaStateObservedEvent("repo-3",
                new ReplicaStateObservation(UNKNOWN, logicalClock.tick(), VERSION)));
        assertEquals(2, registry.countByState(INCONSISTENT));
        assertEquals(1, registry.countByState(UNKNOWN));

        registry.onReplicaStateObserved(new ReplicaStateObservedEvent(REPOSITORY_ID,
                new ReplicaStateObservation(REPAIRING, logicalClock.tick(), VERSION)));
        assertEquals(1, registry.countByState(INCONSISTENT));
        assertEquals(1, registry.countByState(REPAIRING));
        assertEquals(1, registry.countByState(UNKNOWN));

        registry.onReplicaStateObserved(new ReplicaStateObservedEvent(REPOSITORY_ID,
                new ReplicaStateObservation(CONSISTENT, logicalClock.tick(), VERSION)));
        assertEquals(1, registry.countByState(INCONSISTENT));
        assertEquals(0, registry.countByState(REPAIRING));
        assertEquals(1, registry.countByState(UNKNOWN));
    }

    @Test
    public void testEventForOlderVersionButNewerTimestampIsIgnored() {
        //Mark the repository as INCONSISTENT
        registry.onReplicaStateObserved(new ReplicaStateObservedEvent(REPOSITORY_ID,
                new ReplicaStateObservation(INCONSISTENT, timestamp, VERSION)));
        assertEquals(INCONSISTENT, registry.getState(repository));
        verifyStateUpdated(INCONSISTENT, VERSION);

        //Trigger an out-of-order-event, which should be ignored
        registry.onReplicaStateObserved(new ReplicaStateObservedEvent(REPOSITORY_ID,
                new ReplicaStateObservation(CONSISTENT, timestamp + 1L, VERSION - 1L)));
        assertEquals(INCONSISTENT, registry.getState(repository));
        verifyNoMoreInteractions(eventPublisher);
    }

    @Test
    public void testEventForSameVersionButOlderTimestampIsIgnored() {
        //Mark the repository as INCONSISTENT
        registry.onReplicaStateObserved(new ReplicaStateObservedEvent(REPOSITORY_ID,
                new ReplicaStateObservation(INCONSISTENT, timestamp, VERSION)));
        assertEquals(INCONSISTENT, registry.getState(repository));
        verifyStateUpdated(INCONSISTENT, VERSION);

        //Trigger an out-of-order-event, which should be ignored
        registry.onReplicaStateObserved(new ReplicaStateObservedEvent(REPOSITORY_ID,
                new ReplicaStateObservation(CONSISTENT, timestamp - 1L, VERSION)));
        assertEquals(INCONSISTENT, registry.getState(repository));
        verifyNoMoreInteractions(eventPublisher);
    }

    @Test
    public void testForcedEventForOlderVersionButNewerTimestampIsActioned() {
        //Mark the repository as INCONSISTENT
        registry.onReplicaStateObserved(new ReplicaStateObservedEvent(REPOSITORY_ID,
                new ReplicaStateObservation(INCONSISTENT, timestamp, VERSION + 1L)));
        assertEquals(INCONSISTENT, registry.getState(repository));
        verifyStateUpdated(INCONSISTENT, VERSION + 1L);

        //Trigger a forced event that marks the replica as consistent, which should be actioned
        registry.onReplicaStateObserved(new ReplicaStateObservedEvent(REPOSITORY_ID,
                new ReplicaStateObservation(CONSISTENT, timestamp + 1L, VERSION, true)));
        assertEquals(CONSISTENT, registry.getState(repository));
        verifyStateUpdated(CONSISTENT, VERSION);
    }

    @Test
    public void testForcedEventForSameVersionButOlderTimestampIsActioned() {
        //Mark the repository as INCONSISTENT
        registry.onReplicaStateObserved(new ReplicaStateObservedEvent(REPOSITORY_ID,
                new ReplicaStateObservation(INCONSISTENT, timestamp, VERSION)));
        assertEquals(INCONSISTENT, registry.getState(repository));
        verifyStateUpdated(INCONSISTENT, VERSION);

        //Trigger a forced out-of-order-event, which should be actioned
        registry.onReplicaStateObserved(new ReplicaStateObservedEvent(REPOSITORY_ID,
                new ReplicaStateObservation(CONSISTENT, timestamp - 1L, VERSION, true)));
        assertEquals(CONSISTENT, registry.getState(repository));
        verifyStateUpdated(CONSISTENT, VERSION);
    }

    @Test
    public void testGetDetailsForRepository() {
        //If no observation exists for the repository, it is assumed to be consistent at the current version
        assertEquals(new RepositoryReplicaDetails.Builder(CONSISTENT)
                .observedVersion(VERSION)
                .version(VERSION)
                .build(), registry.getDetails(repository));

        //Mark the repository as INCONSISTENT with VERSION + 1
        registry.onReplicaStateObserved(new ReplicaStateObservedEvent(REPOSITORY_ID,
                new ReplicaStateObservation(INCONSISTENT, timestamp, VERSION + 1)));

        assertEquals(new RepositoryReplicaDetails.Builder(INCONSISTENT)
                .observedVersion(VERSION + 1)
                .version(VERSION)
                .build(), registry.getDetails(repository));
    }

    @Test
    public void testInconsistentObservationTriggersStateChange() {
        //Mark the repository as INCONSISTENT
        registry.onReplicaStateObserved(
                new ReplicaStateObservedEvent(REPOSITORY_ID, new ReplicaStateObservation(INCONSISTENT, NOW, VERSION)));
        assertEquals(INCONSISTENT, registry.getState(repository));
        verifyStateUpdated(INCONSISTENT, VERSION);
    }

    @Test
    public void testInconsistentObservationTriggersStateChangeDoesNotTriggerSidebandNotificationWhenAlreadyReported() {
        //Mark the repository as INCONSISTENT
        registry.onReplicaStateObserved(
                new ReplicaStateObservedEvent(REPOSITORY_ID, new ReplicaStateObservation(INCONSISTENT, NOW, VERSION),
                        true));
        assertEquals(INCONSISTENT, registry.getState(repository));
        verifyStateUpdated(INCONSISTENT, VERSION, false);
    }

    @Test
    public void testMarkInconsistentFromApplication() {
        //Mark a REPOSITORY_ID as INCONSISTENT
        registry.onReplicaStateObserved(
                new ReplicaStateObservedEvent(REPOSITORY_ID, new ReplicaStateObservation(INCONSISTENT, NOW, VERSION)));

        reset(eventPublisher, sidebandManager);
        registry.markInconsistent(new MarkInconsistentParameters.Builder()
                .inconsistentReplica("repository-1", 1L)
                .build());

        verify(eventPublisher).publish(eventCaptor.capture());
        //Verify that the control plane is only notified of repositories just marked as inconsistent
        verify(sidebandManager).send(fragmentCaptor.capture());
        assertReplicaObservationFragmentEquals("repository-1", INCONSISTENT, 1L, fragmentCaptor.getValue());
        assertUpdatedEventEquals("repository-1", INCONSISTENT, eventCaptor.getValue());
    }

    @Test
    public void testMarkInconsistentFromControlPlaneComplete() {
        //Mark a few repositories as INCONSISTENT
        registry.onReplicaStateObserved(
                new ReplicaStateObservedEvent("repository-1", new ReplicaStateObservation(INCONSISTENT, NOW - 1, 1L)));
        registry.onReplicaStateObserved(
                new ReplicaStateObservedEvent("repository-2", new ReplicaStateObservation(INCONSISTENT, NOW - 2, 2L)));
        registry.onReplicaStateObserved(
                new ReplicaStateObservedEvent("repository-3", new ReplicaStateObservation(UNKNOWN, NOW - 3, 3L)));
        registry.onReplicaStateObserved(
                new ReplicaStateObservedEvent("repository-4", new ReplicaStateObservation(UNKNOWN, NOW - 4, 4L)));

        reset(eventPublisher, sidebandManager);

        registry.markInconsistent(new MarkInconsistentParameters.Builder()
                .complete(true)
                .fromControlPlane(true)
                .inconsistentReplicas(ImmutableMap.of(
                        "repository-1", 2L, //newer version than currently registered
                        "repository-3", 2L, //older version than currently registered
                        "repository-4", 4L, //same version as currently registered, but registered as UNKNOWN
                        "repository-5", 10L)) //not registered as inconsistent
                .build());

        verify(eventPublisher, times(2)).publish(eventCaptor.capture());
        //First event should be for repository-4, which went from UNKNOWN -> INCONSISTENT
        assertUpdatedEventEquals("repository-4", INCONSISTENT, eventCaptor.getAllValues().get(0));
        //Second event should be for repository-5
        assertUpdatedEventEquals("repository-5", INCONSISTENT, eventCaptor.getAllValues().get(1));

        //Verify that the control plane was only notified of repositories being inconsistent that it did not know
        //about yet: repository-2
        verify(sidebandManager).send(fragmentCaptor.capture());
        assertReplicaObservationFragmentEquals("repository-2", INCONSISTENT, 2L, fragmentCaptor.getValue());
    }

    @Test
    public void testMarkInconsistentFromControlPlaneIncremental() {
        //Mark a few repositories as INCONSISTENT
        registry.onReplicaStateObserved(
                new ReplicaStateObservedEvent("repository-1", new ReplicaStateObservation(INCONSISTENT, NOW - 1, 1L)));
        registry.onReplicaStateObserved(
                new ReplicaStateObservedEvent("repository-2", new ReplicaStateObservation(INCONSISTENT, NOW - 2, 2L)));
        registry.onReplicaStateObserved(
                new ReplicaStateObservedEvent("repository-3", new ReplicaStateObservation(UNKNOWN, NOW - 3, 3L)));
        registry.onReplicaStateObserved(
                new ReplicaStateObservedEvent("repository-4", new ReplicaStateObservation(UNKNOWN, NOW - 4, 4L)));

        reset(eventPublisher, sidebandManager);

        registry.markInconsistent(new MarkInconsistentParameters.Builder()
                .complete(false)
                .fromControlPlane(true)
                .inconsistentReplicas(ImmutableMap.of(
                        "repository-4", 4L, //same version as currently registered, but registered as UNKNOWN
                        "repository-5", 10L)) //not registered as inconsistent
                .build());

        verify(eventPublisher, times(2)).publish(eventCaptor.capture());
        //First event should be for repository-4, which went from UNKNOWN -> INCONSISTENT
        assertUpdatedEventEquals("repository-4", INCONSISTENT, eventCaptor.getAllValues().get(0));
        //Second event should be for repository-5
        assertUpdatedEventEquals("repository-5", INCONSISTENT, eventCaptor.getAllValues().get(1));

        //Verify that the control plane was not notified of any repositories being inconsistent
        verifyNoInteractions(sidebandManager);
    }

    @Test
    public void testMissingEntryAssumedConsistent() {
        assertEquals(CONSISTENT, registry.getState(repository));
    }

    @Test
    public void testOnRepositoryDeletedClearsState() {
        registry.onReplicaStateObserved(
                new ReplicaStateObservedEvent(REPOSITORY_ID, new ReplicaStateObservation(INCONSISTENT, NOW, 1L)));
        assertSame(INCONSISTENT, registry.getState(REPOSITORY_ID));

        registry.onRepositoryDeleted(new RepositoryDeletedEvent(repository));
        assertSame(CONSISTENT, registry.getState(REPOSITORY_ID));
    }

    @Test
    public void testRemove() {
        registry.onReplicaStateObserved(
                new ReplicaStateObservedEvent(REPOSITORY_ID, new ReplicaStateObservation(INCONSISTENT, NOW, 1L)));

        //Removing another repository's state should not remove REPOSITORY_ID's state
        registry.remove("p/0013/h/abcdef52123791279129/r/9732");
        assertSame(INCONSISTENT, registry.getState(REPOSITORY_ID));

        //After removing a repository's replica state, the registry should return CONSISTENT (the default)
        registry.remove(REPOSITORY_ID);
        assertSame(CONSISTENT, registry.getState(REPOSITORY_ID));
    }

    @Test
    public void testUnknownObservationIgnoredIfAlreadyInconsistent() {
        //Mark the repository as INCONSISTENT
        registry.onReplicaStateObserved(new ReplicaStateObservedEvent(REPOSITORY_ID,
                new ReplicaStateObservation(INCONSISTENT, timestamp - 1L, VERSION)));
        assertEquals(INCONSISTENT, registry.getState(repository));
        verifyStateUpdated(INCONSISTENT, VERSION);

        //Trigger an UNKNOWN state observation
        registry.onReplicaStateObserved(new ReplicaStateObservedEvent(REPOSITORY_ID,
                new ReplicaStateObservation(UNKNOWN, timestamp, VERSION)));
        assertEquals(INCONSISTENT, registry.getState(repository));
        verifyNoInteractions(eventPublisher);
    }

    private static void assertReplicaObservationFragmentEquals(String repositoryId, ReplicaState state, long version,
                                                               RpcSidebandFollowerFragment fragment) {
        assertTrue(fragment.hasRepositoryReplicaObservation());

        RpcRepositoryReplicaObservation observation = fragment.getRepositoryReplicaObservation();
        assertEquals(0, observation.getNodesCount()); //Should be a local state observation
        assertEquals(repositoryId, observation.getRepository());
        assertSame(state.toRpc(), observation.getState());
        assertEquals(version, observation.getVersion());
    }

    private static void assertUpdatedEventEquals(String repositoryId, ReplicaState state,
                                                 ReplicaStateUpdatedEvent event) {
        assertNotNull(event);
        assertEquals(repositoryId, event.getRepositoryId());
        assertSame(state, event.getState());
    }

    private void verifyStateUpdated(ReplicaState expected, long version) {
        verifyStateUpdated(expected, version, true);
    }

    private void verifyStateUpdated(ReplicaState expected, long version, boolean sidebandNotification) {
        //Verify the sideband message was sent before the event was published (but both should happen)
        InOrder ordered = inOrder(eventPublisher, sidebandManager);
        if (sidebandNotification) {
            ordered.verify(sidebandManager).send(fragmentCaptor.capture());
        }
        ordered.verify(eventPublisher).publish(eventCaptor.capture());

        if (sidebandNotification) {
            assertReplicaObservationFragmentEquals(REPOSITORY_ID, expected, version, fragmentCaptor.getValue());
        } else {
            verify(sidebandManager, never()).send(any());
        }
        assertUpdatedEventEquals(REPOSITORY_ID, expected, eventCaptor.getValue());

        //Reset mocks so we can verify additional updates
        reset(eventPublisher, sidebandManager);
    }
}