/*
 * Copyright 1999-2023 Alibaba Group Holding Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.alibaba.nacos.naming.consistency.ephemeral.distro.v2;

import com.alibaba.nacos.common.notify.NotifyCenter;
import com.alibaba.nacos.consistency.DataOperation;
import com.alibaba.nacos.core.distributed.distro.DistroProtocol;
import com.alibaba.nacos.core.distributed.distro.entity.DistroData;
import com.alibaba.nacos.core.distributed.distro.entity.DistroKey;
import com.alibaba.nacos.naming.cluster.transport.Serializer;
import com.alibaba.nacos.naming.constants.ClientConstants;
import com.alibaba.nacos.naming.core.v2.ServiceManager;
import com.alibaba.nacos.naming.core.v2.client.Client;
import com.alibaba.nacos.naming.core.v2.client.ClientAttributes;
import com.alibaba.nacos.naming.core.v2.client.ClientSyncData;
import com.alibaba.nacos.naming.core.v2.client.ClientSyncDatumSnapshot;
import com.alibaba.nacos.naming.core.v2.client.impl.ConnectionBasedClient;
import com.alibaba.nacos.naming.core.v2.client.manager.ClientManager;
import com.alibaba.nacos.naming.core.v2.event.client.ClientEvent;
import com.alibaba.nacos.naming.core.v2.pojo.BatchInstanceData;
import com.alibaba.nacos.naming.core.v2.pojo.BatchInstancePublishInfo;
import com.alibaba.nacos.naming.core.v2.pojo.InstancePublishInfo;
import com.alibaba.nacos.naming.core.v2.pojo.Service;
import com.alibaba.nacos.sys.env.EnvUtil;
import com.alibaba.nacos.sys.utils.ApplicationUtils;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.junit.jupiter.MockitoExtension;
import org.mockito.junit.jupiter.MockitoSettings;
import org.mockito.quality.Strictness;
import org.springframework.context.ConfigurableApplicationContext;

import java.util.Arrays;
import java.util.Collections;
import java.util.List;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

@ExtendWith(MockitoExtension.class)
// todo remove this
@MockitoSettings(strictness = Strictness.LENIENT)
class DistroClientDataProcessorTest {
    
    private static final String CLIENT_ID = "11111_1.1.1.1_3306";
    
    private static final String MOCK_TARGET_SERVER = "2.2.2.2:8848";
    
    private Client client;
    
    private DistroData distroData;
    
    private DistroKey distroKey;
    
    private ClientSyncData clientSyncData;
    
    @Mock
    private ClientManager clientManager;
    
    @Mock
    private DistroProtocol distroProtocol;
    
    @Mock
    private ConfigurableApplicationContext applicationContext;
    
    @Mock
    private Serializer serializer;
    
    private DistroClientDataProcessor distroClientDataProcessor;
    
    @BeforeEach
    void setUp() throws Exception {
        distroClientDataProcessor = new DistroClientDataProcessor(clientManager, distroProtocol);
        EnvUtil.setIsStandalone(false);
        client = new ConnectionBasedClient(CLIENT_ID, true, 0L);
        when(clientManager.getClient(CLIENT_ID)).thenReturn(client);
        when(clientManager.isResponsibleClient(client)).thenReturn(true);
        distroData = new DistroData();
        distroKey = new DistroKey();
        distroData.setDistroKey(distroKey);
        distroKey.setTargetServer(MOCK_TARGET_SERVER);
        distroKey.setResourceKey(CLIENT_ID);
        when(applicationContext.getBean(Serializer.class)).thenReturn(serializer);
        ApplicationUtils.injectContext(applicationContext);
        clientSyncData = mockClientSyncData();
        when(serializer.deserialize(any(), eq(ClientSyncData.class))).thenReturn(clientSyncData);
    }
    
    private ClientSyncData mockClientSyncData() {
        ClientSyncData result = new ClientSyncData();
        ClientAttributes clientAttributes = new ClientAttributes();
        clientAttributes.addClientAttribute(ClientConstants.REVISION, 1);
        result.setAttributes(clientAttributes);
        result.setClientId(CLIENT_ID);
        result.setNamespaces(Collections.singletonList("ns"));
        result.setGroupNames(Collections.singletonList("group"));
        result.setServiceNames(Collections.singletonList("service"));
        result.setInstancePublishInfos(Collections.singletonList(new InstancePublishInfo("3.3.3.3", 1111)));
        return result;
    }
    
    @AfterEach
    void tearDown() throws Exception {
        NotifyCenter.deregisterSubscriber(distroClientDataProcessor);
    }
    
    @Test
    void testFinishInitial() {
        assertFalse(distroClientDataProcessor.isFinishInitial());
        distroClientDataProcessor.finishInitial();
        assertTrue(distroClientDataProcessor.isFinishInitial());
    }
    
    @Test
    void processType() {
        assertEquals(DistroClientDataProcessor.TYPE, distroClientDataProcessor.processType());
    }
    
    @Test
    void testOnEventForStandalone() {
        EnvUtil.setIsStandalone(true);
        distroClientDataProcessor.onEvent(new ClientEvent.ClientVerifyFailedEvent(CLIENT_ID, MOCK_TARGET_SERVER));
        verify(distroProtocol, never()).syncToTarget(any(), any(), anyString(), anyLong());
        verify(distroProtocol, never()).sync(any(), any());
    }
    
    @Test
    void testOnClientVerifyFailedEventWithoutClient() {
        when(clientManager.getClient(CLIENT_ID)).thenReturn(null);
        distroClientDataProcessor.onEvent(new ClientEvent.ClientVerifyFailedEvent(CLIENT_ID, MOCK_TARGET_SERVER));
        verify(distroProtocol, never()).syncToTarget(any(), any(), anyString(), anyLong());
        verify(distroProtocol, never()).sync(any(), any());
    }
    
    @Test
    void testOnClientVerifyFailedEventWithPersistentClient() {
        client = mock(Client.class);
        when(client.isEphemeral()).thenReturn(false);
        when(clientManager.getClient(CLIENT_ID)).thenReturn(client);
        distroClientDataProcessor.onEvent(new ClientEvent.ClientVerifyFailedEvent(CLIENT_ID, MOCK_TARGET_SERVER));
        verify(distroProtocol, never()).syncToTarget(any(), any(), anyString(), anyLong());
        verify(distroProtocol, never()).sync(any(), any());
    }
    
    @Test
    void testOnClientVerifyFailedEventWithoutResponsible() {
        when(clientManager.isResponsibleClient(client)).thenReturn(false);
        distroClientDataProcessor.onEvent(new ClientEvent.ClientVerifyFailedEvent(CLIENT_ID, MOCK_TARGET_SERVER));
        verify(distroProtocol, never()).syncToTarget(any(), any(), anyString(), anyLong());
        verify(distroProtocol, never()).sync(any(), any());
    }
    
    @Test
    void testOnClientVerifyFailedEventSuccess() {
        distroClientDataProcessor.onEvent(new ClientEvent.ClientVerifyFailedEvent(CLIENT_ID, MOCK_TARGET_SERVER));
        verify(distroProtocol).syncToTarget(any(), eq(DataOperation.ADD), eq(MOCK_TARGET_SERVER), eq(0L));
        verify(distroProtocol, never()).sync(any(), any());
    }
    
    @Test
    void testOnClientChangedEventWithoutClient() {
        distroClientDataProcessor.onEvent(new ClientEvent.ClientChangedEvent(null));
        verify(distroProtocol, never()).syncToTarget(any(), any(), anyString(), anyLong());
        verify(distroProtocol, never()).sync(any(), any());
    }
    
    @Test
    void testOnClientChangedEventWithPersistentClient() {
        client = mock(Client.class);
        when(client.isEphemeral()).thenReturn(false);
        distroClientDataProcessor.onEvent(new ClientEvent.ClientChangedEvent(client));
        verify(distroProtocol, never()).syncToTarget(any(), any(), anyString(), anyLong());
        verify(distroProtocol, never()).sync(any(), any());
    }
    
    @Test
    void testOnClientChangedEventWithoutResponsible() {
        when(clientManager.isResponsibleClient(client)).thenReturn(false);
        distroClientDataProcessor.onEvent(new ClientEvent.ClientChangedEvent(client));
        verify(distroProtocol, never()).syncToTarget(any(), any(), anyString(), anyLong());
        verify(distroProtocol, never()).sync(any(), any());
    }
    
    @Test
    void testOnClientChangedEventSuccess() {
        distroClientDataProcessor.onEvent(new ClientEvent.ClientChangedEvent(client));
        verify(distroProtocol, never()).syncToTarget(any(), any(), anyString(), anyLong());
        verify(distroProtocol).sync(any(), eq(DataOperation.CHANGE));
    }
    
    @Test
    void testOnClientDisconnectEventSuccess() {
        distroClientDataProcessor.onEvent(new ClientEvent.ClientDisconnectEvent(client, true));
        verify(distroProtocol, never()).syncToTarget(any(), any(), anyString(), anyLong());
        verify(distroProtocol).sync(any(), eq(DataOperation.DELETE));
    }
    
    @Test
    void testProcessDataForDeleteClient() {
        distroData.setType(DataOperation.DELETE);
        distroClientDataProcessor.processData(distroData);
        verify(clientManager).clientDisconnected(CLIENT_ID);
    }
    
    @Test
    void testProcessDataForChangeClient() {
        distroData.setType(DataOperation.CHANGE);
        assertEquals(0L, client.getRevision());
        assertEquals(0, client.getAllPublishedService().size());
        distroClientDataProcessor.processData(distroData);
        verify(clientManager).syncClientConnected(CLIENT_ID, clientSyncData.getAttributes());
        assertEquals(1L, client.getRevision());
        assertEquals(1, client.getAllPublishedService().size());
    }
    
    @Test
    void testProcessDataForBatch() {
        // swap tmp
        Serializer mock = Mockito.mock(Serializer.class);
        when(applicationContext.getBean(Serializer.class)).thenReturn(mock);
        
        // single instance => batch instances => batch instances => single instance
        // single
        ClientSyncData syncData = createSingleForBatchTest(1);
        DistroData data = new DistroData();
        data.setContent(serializer.serialize(syncData));
        data.setType(DataOperation.ADD);
        when(mock.deserialize(any(), eq(ClientSyncData.class))).thenReturn(syncData);
        distroClientDataProcessor.processData(data);
        assertEquals(1L, client.getRevision());
        assertEquals(1, client.getAllPublishedService().size());
        Service service = Service.newService("batchData", "batchData", "batchData");
        Service singleton = ServiceManager.getInstance().getSingleton(service);
        InstancePublishInfo info = client.getInstancePublishInfo(ServiceManager.getInstance().getSingleton(singleton));
        assertEquals("127.0.0.1", info.getIp());
        assertEquals(8080, info.getPort());
        
        // batch
        data = new DistroData();
        syncData = createBatchForBatchTest(2);
        data.setContent(serializer.serialize(syncData));
        data.setType(DataOperation.CHANGE);
        when(mock.deserialize(any(), eq(ClientSyncData.class))).thenReturn(syncData);
        distroClientDataProcessor.processData(data);
        assertEquals(2L, client.getRevision());
        assertEquals(1, client.getAllPublishedService().size());
        info = client.getInstancePublishInfo(ServiceManager.getInstance().getSingleton(singleton));
        assertTrue(info instanceof BatchInstancePublishInfo);
        BatchInstancePublishInfo batchInfo = (BatchInstancePublishInfo) info;
        assertEquals(2, batchInfo.getInstancePublishInfos().size());
        for (InstancePublishInfo instancePublishInfo : batchInfo.getInstancePublishInfos()) {
            assertEquals("127.0.0.1", instancePublishInfo.getIp());
            assertTrue(instancePublishInfo.getPort() == 8080 || instancePublishInfo.getPort() == 8081);
        }
        
        // batch
        data = new DistroData();
        syncData = createBatchForBatchTest(3);
        data.setContent(serializer.serialize(syncData));
        data.setType(DataOperation.CHANGE);
        when(mock.deserialize(any(), eq(ClientSyncData.class))).thenReturn(syncData);
        distroClientDataProcessor.processData(data);
        assertEquals(3L, client.getRevision());
        assertEquals(1, client.getAllPublishedService().size());
        info = client.getInstancePublishInfo(ServiceManager.getInstance().getSingleton(singleton));
        assertTrue(info instanceof BatchInstancePublishInfo);
        batchInfo = (BatchInstancePublishInfo) info;
        assertEquals(2, batchInfo.getInstancePublishInfos().size());
        for (InstancePublishInfo instancePublishInfo : batchInfo.getInstancePublishInfos()) {
            assertEquals("127.0.0.1", instancePublishInfo.getIp());
            assertTrue(instancePublishInfo.getPort() == 8080 || instancePublishInfo.getPort() == 8081);
        }
        
        // single
        syncData = createSingleForBatchTest(4);
        data = new DistroData();
        data.setContent(serializer.serialize(syncData));
        data.setType(DataOperation.ADD);
        when(mock.deserialize(any(), eq(ClientSyncData.class))).thenReturn(syncData);
        distroClientDataProcessor.processData(data);
        assertEquals(4L, client.getRevision());
        assertEquals(1, client.getAllPublishedService().size());
        info = client.getInstancePublishInfo(ServiceManager.getInstance().getSingleton(singleton));
        assertEquals("127.0.0.1", info.getIp());
        assertEquals(8080, info.getPort());
    }
    
    private ClientSyncData createSingleForBatchTest(int revision) {
        ClientSyncData syncData = new ClientSyncData();
        syncData.setClientId(CLIENT_ID);
        ClientAttributes clientAttributes = new ClientAttributes();
        clientAttributes.addClientAttribute(ClientConstants.REVISION, revision);
        syncData.setAttributes(clientAttributes);
        syncData.setNamespaces(Collections.singletonList("batchData"));
        syncData.setGroupNames(Collections.singletonList("batchData"));
        syncData.setServiceNames(Collections.singletonList("batchData"));
        syncData.setInstancePublishInfos(Collections.singletonList(new InstancePublishInfo("127.0.0.1", 8080)));
        return syncData;
    }
    
    private ClientSyncData createBatchForBatchTest(int revision) {
        ClientSyncData syncData = new ClientSyncData();
        syncData.setClientId(CLIENT_ID);
        ClientAttributes clientAttributes = new ClientAttributes();
        clientAttributes.addClientAttribute(ClientConstants.REVISION, revision);
        syncData.setAttributes(clientAttributes);
        syncData.setNamespaces(Collections.emptyList());
        BatchInstancePublishInfo batchInstancePublishInfo = new BatchInstancePublishInfo();
        syncData.setBatchInstanceData(new BatchInstanceData(Collections.singletonList("batchData"), Collections.singletonList("batchData"),
                Collections.singletonList("batchData"), Collections.singletonList(batchInstancePublishInfo)));
        batchInstancePublishInfo.setInstancePublishInfos(
                Arrays.asList(new InstancePublishInfo("127.0.0.1", 8080), new InstancePublishInfo("127.0.0.1", 8081)));
        return syncData;
    }
    
    @Test
    void testProcessVerifyData() {
        DistroClientVerifyInfo verifyInfo = new DistroClientVerifyInfo(CLIENT_ID, 0L);
        when(serializer.deserialize(any(), eq(DistroClientVerifyInfo.class))).thenReturn(verifyInfo);
        assertFalse(distroClientDataProcessor.processVerifyData(distroData, MOCK_TARGET_SERVER));
        when(clientManager.verifyClient(verifyInfo)).thenReturn(true);
        assertTrue(distroClientDataProcessor.processVerifyData(distroData, MOCK_TARGET_SERVER));
    }
    
    @Test
    void testProcessSnapshot() {
        ClientSyncDatumSnapshot snapshot = new ClientSyncDatumSnapshot();
        snapshot.setClientSyncDataList(Collections.singletonList(clientSyncData));
        when(serializer.deserialize(any(), eq(ClientSyncDatumSnapshot.class))).thenReturn(snapshot);
        assertEquals(0L, client.getRevision());
        assertEquals(0, client.getAllPublishedService().size());
        distroClientDataProcessor.processSnapshot(distroData);
        verify(clientManager).syncClientConnected(CLIENT_ID, clientSyncData.getAttributes());
        assertEquals(1L, client.getRevision());
        assertEquals(1, client.getAllPublishedService().size());
    }
    
    @Test
    void testGetDistroData() {
        DistroData actual = distroClientDataProcessor.getDistroData(distroKey);
        assertEquals(distroKey, actual.getDistroKey());
    }
    
    @Test
    void testGetDatumSnapshot() {
        when(clientManager.allClientId()).thenReturn(Collections.singletonList(CLIENT_ID));
        DistroData actual = distroClientDataProcessor.getDatumSnapshot();
        assertEquals(DataOperation.SNAPSHOT.name(), actual.getDistroKey().getResourceKey());
        assertEquals(DistroClientDataProcessor.TYPE, actual.getDistroKey().getResourceType());
    }
    
    @Test
    void testGetVerifyData() {
        client.setRevision(10L);
        when(clientManager.allClientId()).thenReturn(Collections.singletonList(CLIENT_ID));
        List<DistroData> list = distroClientDataProcessor.getVerifyData();
        assertEquals(1, list.size());
        assertEquals(DataOperation.VERIFY, list.iterator().next().getType());
        assertEquals(CLIENT_ID, list.iterator().next().getDistroKey().getResourceKey());
        assertEquals(DistroClientDataProcessor.TYPE, list.iterator().next().getDistroKey().getResourceType());
    }
}