/*
 * Copyright 1999-2018 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.client.config.impl;

import com.alibaba.nacos.api.ability.constant.AbilityKey;
import com.alibaba.nacos.api.config.listener.AbstractFuzzyWatchEventWatcher;
import com.alibaba.nacos.api.config.listener.ConfigFuzzyWatchChangeEvent;
import com.alibaba.nacos.api.config.remote.request.ConfigFuzzyWatchChangeNotifyRequest;
import com.alibaba.nacos.api.config.remote.request.ConfigFuzzyWatchRequest;
import com.alibaba.nacos.api.config.remote.request.ConfigFuzzyWatchSyncRequest;
import com.alibaba.nacos.api.config.remote.response.ConfigFuzzyWatchResponse;
import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.exception.runtime.NacosRuntimeException;
import com.alibaba.nacos.client.config.common.GroupKey;
import com.alibaba.nacos.common.remote.client.RpcClient;
import com.alibaba.nacos.common.utils.CollectionUtils;
import com.alibaba.nacos.common.utils.FuzzyGroupKeyPattern;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.ArgumentMatchers;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.junit.jupiter.MockitoExtension;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import static com.alibaba.nacos.api.common.Constants.ConfigChangedType.ADD_CONFIG;
import static com.alibaba.nacos.api.common.Constants.ConfigChangedType.DELETE_CONFIG;
import static com.alibaba.nacos.api.common.Constants.FUZZY_WATCH_DIFF_SYNC_NOTIFY;
import static com.alibaba.nacos.api.common.Constants.FUZZY_WATCH_INIT_NOTIFY;
import static com.alibaba.nacos.api.model.v2.ErrorCode.FUZZY_WATCH_PATTERN_MATCH_COUNT_OVER_LIMIT;
import static com.alibaba.nacos.api.model.v2.ErrorCode.FUZZY_WATCH_PATTERN_OVER_LIMIT;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.reset;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

@ExtendWith(MockitoExtension.class)
public class ConfigFuzzyWatchGroupKeyHolderTest {
    
    ConfigFuzzyWatchGroupKeyHolder configFuzzyWatchGroupKeyHolder;
    
    @Mock
    ClientWorker.ConfigRpcTransportClient rpcTransportClient;
    
    String clientId = "conn" + System.currentTimeMillis();
    
    String tenant = "t1";
    
    @BeforeEach
    void before() {
        configFuzzyWatchGroupKeyHolder = new ConfigFuzzyWatchGroupKeyHolder(rpcTransportClient, clientId);
        doReturn(true).when(rpcTransportClient).isAbilitySupportedByServer(AbilityKey.SERVER_FUZZY_WATCH);
    }
    
    @AfterEach
    void after() {
    }
    
    @Test
    void testRegisterFuzzyWatcherAndNotify() throws InterruptedException {
        when(rpcTransportClient.getTenant()).thenReturn(tenant);
        
        String dataId = "dataId";
        String group = "group";
        
        AtomicInteger watcher1Flag = new AtomicInteger(0);
        AtomicInteger watcher2Flag = new AtomicInteger(0);
        configFuzzyWatchGroupKeyHolder.registerFuzzyWatcher(dataId + "*", group, new AbstractFuzzyWatchEventWatcher() {
            @Override
            public void onEvent(ConfigFuzzyWatchChangeEvent event) {
                watcher1Flag.incrementAndGet();
            }
        });
        
        configFuzzyWatchGroupKeyHolder.registerFuzzyWatcher(dataId + "*", group, new AbstractFuzzyWatchEventWatcher() {
            @Override
            public void onEvent(ConfigFuzzyWatchChangeEvent event) {
                watcher2Flag.incrementAndGet();
            }
        });
        
        String groupKey1 = GroupKey.getKeyTenant(dataId + 1, group, tenant);
        //build init notify add
        Set<ConfigFuzzyWatchSyncRequest.Context> contexts = new HashSet<>();
        contexts.add(ConfigFuzzyWatchSyncRequest.Context.build(groupKey1, ADD_CONFIG));
        String groupKeyPattern = FuzzyGroupKeyPattern.generatePattern(dataId + "*", group, tenant);
        ConfigFuzzyWatchSyncRequest initNotifyRequest = ConfigFuzzyWatchSyncRequest.buildSyncRequest(
                FUZZY_WATCH_INIT_NOTIFY, contexts, groupKeyPattern, 1, 1);
        configFuzzyWatchGroupKeyHolder.handleFuzzyWatchSyncNotifyRequest(initNotifyRequest);
        //check watcher notified
        Thread.sleep(100L);
        Assertions.assertTrue(watcher1Flag.get() == 1);
        Assertions.assertTrue(watcher1Flag.get() == 1);
        
        //build change notify add
        String changedGroupKey2Add = GroupKey.getKeyTenant(dataId + 2, group, tenant);
        ConfigFuzzyWatchChangeNotifyRequest changedNotifyRequest = new ConfigFuzzyWatchChangeNotifyRequest(
                changedGroupKey2Add, ADD_CONFIG);
        configFuzzyWatchGroupKeyHolder.handlerFuzzyWatchChangeNotifyRequest(changedNotifyRequest);
        
        //check watcher notified
        Thread.sleep(100L);
        Assertions.assertTrue(watcher1Flag.get() == 2);
        Assertions.assertTrue(watcher1Flag.get() == 2);
        
        //check not complete future timeout
        ConfigFuzzyWatchContext configFuzzyWatchContext = configFuzzyWatchGroupKeyHolder.getFuzzyListenContext(
                dataId + "*", group);
        Future<Set<String>> newFutureNotFinish = configFuzzyWatchContext.createNewFuture();
        try {
            newFutureNotFinish.get(1000L, TimeUnit.MILLISECONDS);
            Assertions.assertFalse(true);
        } catch (TimeoutException e) {
            Assertions.assertTrue(true);
        } catch (Throwable throwable) {
            Assertions.assertFalse(true);
        }
        
        // build init finish notify
        ConfigFuzzyWatchSyncRequest configFuzzyWatchSyncRequest = ConfigFuzzyWatchSyncRequest.buildInitFinishRequest(
                groupKeyPattern);
        configFuzzyWatchGroupKeyHolder.handleFuzzyWatchSyncNotifyRequest(configFuzzyWatchSyncRequest);
        
        //check a completed future.
        Future<Set<String>> newFutureFinish = configFuzzyWatchContext.createNewFuture();
        try {
            Set<String> groupKeys = newFutureFinish.get(10L, TimeUnit.MILLISECONDS);
            Assertions.assertTrue(
                    groupKeys != null && groupKeys.contains(groupKey1) && groupKeys.contains(changedGroupKey2Add));
        } catch (Exception e) {
            Assertions.assertTrue(false);
        }
        
        //check watcher notified delete
        Thread.sleep(100L);
        Assertions.assertTrue(watcher1Flag.get() == 2);
        Assertions.assertTrue(watcher1Flag.get() == 2);
        
        //build change notify
        String changedGroupKey2Delete = changedGroupKey2Add;
        
        ConfigFuzzyWatchChangeNotifyRequest changedNotifyRequestDelete = new ConfigFuzzyWatchChangeNotifyRequest(
                changedGroupKey2Delete, DELETE_CONFIG);
        configFuzzyWatchGroupKeyHolder.handlerFuzzyWatchChangeNotifyRequest(changedNotifyRequestDelete);
        //check watcher notified delete
        Thread.sleep(100L);
        Assertions.assertTrue(watcher1Flag.get() == 3);
        Assertions.assertTrue(watcher1Flag.get() == 3);
        
        Future<Set<String>> newFuture = configFuzzyWatchContext.createNewFuture();
        
        try {
            Set<String> groupKeys = newFuture.get(10L, TimeUnit.MILLISECONDS);
            Assertions.assertTrue(
                    groupKeys != null && groupKeys.contains(groupKey1) && !groupKeys.contains(changedGroupKey2Delete));
        } catch (Exception e) {
            Assertions.assertTrue(false);
        }
        
        //build sync delete
        String groupKey1Delete = groupKey1;
        //build init notify add
        Set<ConfigFuzzyWatchSyncRequest.Context> contextsDelete = new HashSet<>();
        contextsDelete.add(ConfigFuzzyWatchSyncRequest.Context.build(groupKey1Delete, DELETE_CONFIG));
        ConfigFuzzyWatchSyncRequest deleteNotifyRequest = ConfigFuzzyWatchSyncRequest.buildSyncRequest(
                FUZZY_WATCH_DIFF_SYNC_NOTIFY, contextsDelete, groupKeyPattern, 1, 1);
        configFuzzyWatchGroupKeyHolder.handleFuzzyWatchSyncNotifyRequest(deleteNotifyRequest);
        
        //check watcher notified delete
        Thread.sleep(100L);
        Assertions.assertTrue(watcher1Flag.get() == 4);
        Assertions.assertTrue(watcher1Flag.get() == 4);
        
        Future<Set<String>> newFutureEmpty = configFuzzyWatchContext.createNewFuture();
        
        try {
            Set<String> groupKeys = newFutureEmpty.get(10L, TimeUnit.MILLISECONDS);
            Assertions.assertTrue(CollectionUtils.isEmpty(groupKeys));
        } catch (Exception e) {
            Assertions.assertTrue(false);
        }
        
        configFuzzyWatchGroupKeyHolder.resetConsistenceStatus();
        Assertions.assertFalse(configFuzzyWatchContext.isConsistentWithServer());
    }
    
    @Test
    void testExecuteConfigFuzzyListen() throws NacosException {
        when(rpcTransportClient.getTenant()).thenReturn(tenant);
        
        ConfigFuzzyWatchContext configFuzzyWatchContext = configFuzzyWatchGroupKeyHolder.registerFuzzyWatcher("da1*",
                "group*", new AbstractFuzzyWatchEventWatcher() {
                    @Override
                    public void onEvent(ConfigFuzzyWatchChangeEvent event) {
                    
                    }
                });
        configFuzzyWatchContext.setConsistentWithServer(true);
        
        configFuzzyWatchGroupKeyHolder.registerFuzzyWatcher("da2*", "group*", new AbstractFuzzyWatchEventWatcher() {
            @Override
            public void onEvent(ConfigFuzzyWatchChangeEvent event) {
            
            }
        });
        
        configFuzzyWatchGroupKeyHolder.registerFuzzyWatcher("da3*", "group*", new AbstractFuzzyWatchEventWatcher() {
            @Override
            public void onEvent(ConfigFuzzyWatchChangeEvent event) {
            
            }
        });
        
        RpcClient rpcClient = Mockito.mock(RpcClient.class);
        when(rpcTransportClient.ensureRpcClient(eq("0"))).thenReturn(rpcClient);
        ScheduledExecutorService scheduledExecutorService = Mockito.mock(ScheduledExecutorService.class);
        when(rpcTransportClient.getExecutor()).thenReturn(scheduledExecutorService);
        when(scheduledExecutorService.submit(any(Runnable.class))).thenReturn(Mockito.mock(Future.class));
        configFuzzyWatchGroupKeyHolder.executeConfigFuzzyListen();
        
        verify(scheduledExecutorService, times(2)).submit(ArgumentMatchers.any(Runnable.class));
        
    }
    
    @Test
    void testExecuteFuzzyWatchRequestNormal() throws NacosException {
        reset(rpcTransportClient);
        String envName = "name";
        String groupKeyPattern = "pattern";
        ConfigFuzzyWatchContext configFuzzyWatchContext = new ConfigFuzzyWatchContext(envName, groupKeyPattern);
        configFuzzyWatchContext.refreshOverLimitTs();
        RpcClient rpcClient = Mockito.mock(RpcClient.class);
        
        when(rpcTransportClient.requestProxy(eq(rpcClient), any(ConfigFuzzyWatchRequest.class))).thenReturn(
                new ConfigFuzzyWatchResponse());
        configFuzzyWatchGroupKeyHolder.executeFuzzyWatchRequest(configFuzzyWatchContext, rpcClient);
        
        Assertions.assertTrue(!configFuzzyWatchContext.patternLimitSuppressed());
        Assertions.assertTrue(configFuzzyWatchContext.isConsistentWithServer());
        
    }
    
    @Test
    void testExecuteFuzzyWatchRequestRemove() throws NacosException {
        
        AbstractFuzzyWatchEventWatcher abstractFuzzyWatchEventWatcher = new AbstractFuzzyWatchEventWatcher() {
            @Override
            public void onEvent(ConfigFuzzyWatchChangeEvent event) {
            
            }
        };
        configFuzzyWatchGroupKeyHolder.registerFuzzyWatcher("*", "*", abstractFuzzyWatchEventWatcher);
        configFuzzyWatchGroupKeyHolder.removeFuzzyWatcher("*", "*", abstractFuzzyWatchEventWatcher);
        
        RpcClient rpcClient = Mockito.mock(RpcClient.class);
        when(rpcTransportClient.requestProxy(eq(rpcClient), any(ConfigFuzzyWatchRequest.class))).thenReturn(
                new ConfigFuzzyWatchResponse());
        configFuzzyWatchGroupKeyHolder.executeFuzzyWatchRequest(
                configFuzzyWatchGroupKeyHolder.getFuzzyListenContext("*", "*"), rpcClient);
        
        Assertions.assertTrue(configFuzzyWatchGroupKeyHolder.getFuzzyListenContext("*", "*") == null);
        
    }
    
    @Test
    void testExecuteFuzzyWatchRequestOverLoad() throws NacosException, InterruptedException {
        
        AtomicBoolean patternOvrFlag = new AtomicBoolean(false);
        AtomicBoolean configCountOverFlag = new AtomicBoolean(false);
        
        AbstractFuzzyWatchEventWatcher abstractFuzzyWatchEventWatcher = new AbstractFuzzyWatchEventWatcher() {
            @Override
            public void onEvent(ConfigFuzzyWatchChangeEvent event) {
            
            }
            
            @Override
            public void onPatternOverLimit() {
                patternOvrFlag.set(true);
            }
            
            @Override
            public void onConfigReachUpLimit() {
                configCountOverFlag.set(true);
            }
        };
        ConfigFuzzyWatchContext configFuzzyWatchContext = configFuzzyWatchGroupKeyHolder.registerFuzzyWatcher("*", "*",
                abstractFuzzyWatchEventWatcher);
        
        RpcClient rpcClient = Mockito.mock(RpcClient.class);
        
        //test pattern over load
        ConfigFuzzyWatchResponse overloadResponse = new ConfigFuzzyWatchResponse();
        overloadResponse.setErrorInfo(FUZZY_WATCH_PATTERN_OVER_LIMIT.getCode(),
                FUZZY_WATCH_PATTERN_OVER_LIMIT.getMsg());
        when(rpcTransportClient.requestProxy(eq(rpcClient), any(ConfigFuzzyWatchRequest.class))).thenReturn(
                overloadResponse);
        configFuzzyWatchGroupKeyHolder.executeFuzzyWatchRequest(configFuzzyWatchContext, rpcClient);
        Thread.sleep(100L);
        Assertions.assertTrue(configFuzzyWatchContext.patternLimitSuppressed());
        Assertions.assertTrue(!configFuzzyWatchContext.isConsistentWithServer());
        Assertions.assertTrue(patternOvrFlag.get());
        Assertions.assertFalse(configCountOverFlag.get());
        
        configFuzzyWatchContext.clearOverLimitTs();
        ConfigFuzzyWatchResponse countOverloadResponse = new ConfigFuzzyWatchResponse();
        countOverloadResponse.setErrorInfo(FUZZY_WATCH_PATTERN_MATCH_COUNT_OVER_LIMIT.getCode(),
                FUZZY_WATCH_PATTERN_MATCH_COUNT_OVER_LIMIT.getMsg());
        when(rpcTransportClient.requestProxy(eq(rpcClient), any(ConfigFuzzyWatchRequest.class))).thenReturn(
                countOverloadResponse);
        configFuzzyWatchGroupKeyHolder.executeFuzzyWatchRequest(configFuzzyWatchContext, rpcClient);
        Thread.sleep(100L);
        Assertions.assertTrue(configFuzzyWatchContext.patternLimitSuppressed());
        Assertions.assertTrue(!configFuzzyWatchContext.isConsistentWithServer());
        Assertions.assertTrue(configCountOverFlag.get());
        
    }
    
    @Test
    void testSyncWhenWatcherFail() throws NacosException, InterruptedException {
        when(rpcTransportClient.getTenant()).thenReturn(tenant);
        
        String groupKey = GroupKey.getKeyTenant("dataIdName124", "group", tenant);
        
        AtomicInteger watcherFlag = new AtomicInteger(0);
        
        AbstractFuzzyWatchEventWatcher abstractFuzzyWatchEventWatcher = new AbstractFuzzyWatchEventWatcher() {
            
            @Override
            public void onEvent(ConfigFuzzyWatchChangeEvent event) {
                int get = watcherFlag.incrementAndGet();
                if (get < 2) {
                    System.out.println("times " + get + " fail");
                    throw new RuntimeException("mock exception");
                } else {
                    System.out.println("times " + get + " success");
                    
                }
            }
        };
        
        ConfigFuzzyWatchContext configFuzzyWatchContext = configFuzzyWatchGroupKeyHolder.registerFuzzyWatcher(
                "dataIdName*", "group", abstractFuzzyWatchEventWatcher);
        
        ConfigFuzzyWatchChangeNotifyRequest configFuzzyWatchChangeNotifyRequest = new ConfigFuzzyWatchChangeNotifyRequest(
                groupKey, ADD_CONFIG);
        configFuzzyWatchGroupKeyHolder.handlerFuzzyWatchChangeNotifyRequest(configFuzzyWatchChangeNotifyRequest);
        
        TimeUnit.MILLISECONDS.sleep(100L);
        
        //notify 1, fail
        configFuzzyWatchContext.syncFuzzyWatchers();
        //notify 2,success
        configFuzzyWatchContext.syncFuzzyWatchers();
        //notify 3 +, will not trigger watchers.
        configFuzzyWatchContext.syncFuzzyWatchers();
        configFuzzyWatchContext.syncFuzzyWatchers();
        configFuzzyWatchContext.syncFuzzyWatchers();
        //expect  2 times notified
        Assertions.assertEquals(2, watcherFlag.get());
    }
    
    @Test
    void testFuzzyWatchNotSupport() {
        when(rpcTransportClient.isAbilitySupportedByServer(AbilityKey.SERVER_FUZZY_WATCH)).thenReturn(false);
        Assertions.assertThrows(NacosRuntimeException.class, () -> {
            configFuzzyWatchGroupKeyHolder.registerFuzzyWatcher("dataIdName*", "group",
                    new AbstractFuzzyWatchEventWatcher() {
                        
                        @Override
                        public void onEvent(ConfigFuzzyWatchChangeEvent event) {
                        }
                    });
        });
    }
}
