/*
 * 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.client.naming.cache;

import com.alibaba.nacos.api.ability.constant.AbilityKey;
import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.exception.runtime.NacosRuntimeException;
import com.alibaba.nacos.api.naming.listener.FuzzyWatchEventWatcher;
import com.alibaba.nacos.api.naming.remote.request.NamingFuzzyWatchRequest;
import com.alibaba.nacos.api.naming.remote.response.NamingFuzzyWatchResponse;
import com.alibaba.nacos.client.naming.event.NamingFuzzyWatchLoadEvent;
import com.alibaba.nacos.client.naming.event.NamingFuzzyWatchNotifyEvent;
import com.alibaba.nacos.client.naming.remote.gprc.NamingGrpcClientProxy;
import com.alibaba.nacos.client.utils.LogUtils;
import com.alibaba.nacos.common.executor.NameThreadFactory;
import com.alibaba.nacos.common.notify.Event;
import com.alibaba.nacos.common.notify.NotifyCenter;
import com.alibaba.nacos.common.notify.listener.SmartSubscriber;
import com.alibaba.nacos.common.utils.CollectionUtils;
import org.slf4j.Logger;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import static com.alibaba.nacos.api.common.Constants.FUZZY_WATCH_INIT_NOTIFY;
import static com.alibaba.nacos.api.common.Constants.ServiceChangedType.ADD_SERVICE;
import static com.alibaba.nacos.api.common.Constants.WATCH_TYPE_CANCEL_WATCH;
import static com.alibaba.nacos.api.common.Constants.WATCH_TYPE_WATCH;
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;

/**
 * Naming client fuzzy watch service list holder.
 *
 * @author tanyongquan
 */
public class NamingFuzzyWatchServiceListHolder extends SmartSubscriber {
    
    private static final Logger LOGGER = LogUtils.logger(NamingFuzzyWatchServiceListHolder.class);
    
    private String notifierEventScope;
    
    private NamingGrpcClientProxy namingGrpcClientProxy;
    
    /**
     * fuzzyListenExecuteBell.
     */
    private final BlockingQueue<Object> fuzzyWatchExecuteBell = new ArrayBlockingQueue<>(1);
    
    private final Object bellItem = new Object();
    
    private final AtomicLong fuzzyWatchLastAllSyncTime = new AtomicLong(System.currentTimeMillis());
    
    private static final long FUZZY_LISTEN_ALL_SYNC_INTERNAL = 3 * 60 * 1000;
    
    ScheduledExecutorService executorService;
    
    /**
     * The contents of {@code patternMatchMap} are Map{pattern -> Set[matched services]}.
     */
    private Map<String, NamingFuzzyWatchContext> fuzzyMatchContextMap = new ConcurrentHashMap<>();
    
    public NamingFuzzyWatchServiceListHolder(String notifierEventScope) {
        this.notifierEventScope = notifierEventScope;
        NotifyCenter.registerSubscriber(this);
    }
    
    /**
     * shut down.
     */
    public void shutdown() {
        if (executorService != null && !executorService.isShutdown()) {
            executorService.shutdown();
        }
    }
    
    /**
     * start.
     */
    @SuppressWarnings("PMD.ThreadPoolCreationRule")
    public void start() {
        
        executorService = Executors.newSingleThreadScheduledExecutor(
                new NameThreadFactory("com.alibaba.nacos.client.naming.fuzzy.watch.Worker"));
        executorService.submit(() -> {
            while (!executorService.isShutdown() && !executorService.isTerminated()) {
                try {
                    fuzzyWatchExecuteBell.poll(5L, TimeUnit.SECONDS);
                    if (executorService.isShutdown() || executorService.isTerminated()) {
                        continue;
                    }
                    executeNamingFuzzyWatch();
                } catch (Throwable e) {
                    LOGGER.error("[rpc-fuzzy-watch-execute] rpc fuzzy watch exception", e);
                    try {
                        Thread.sleep(50L);
                    } catch (InterruptedException interruptedException) {
                        //ignore
                    }
                    notifyFuzzyWatchSync();
                }
            }
        });
    }
    
    public void registerNamingGrpcClientProxy(NamingGrpcClientProxy namingGrpcClientProxy) {
        this.namingGrpcClientProxy = namingGrpcClientProxy;
    }
    
    public NamingFuzzyWatchContext getFuzzyWatchContext(String groupKeyPattern) {
        return fuzzyMatchContextMap.get(groupKeyPattern);
    }
    
    /**
     * Add a watcher to the context.
     *
     * @param watcher watcher to be added
     */
    public NamingFuzzyWatchContext registerFuzzyWatcher(String groupKeyPattern, FuzzyWatchEventWatcher watcher) {
        if (!namingGrpcClientProxy.isAbilitySupportedByServer(AbilityKey.SERVER_FUZZY_WATCH)) {
            throw new NacosRuntimeException(NacosException.SERVER_NOT_IMPLEMENTED,
                    "Request Nacos server version is too low, not support fuzzy watch feature.");
        }
        NamingFuzzyWatchContext namingFuzzyWatchContext = initFuzzyWatchContextIfNeed(groupKeyPattern);
        namingFuzzyWatchContext.setDiscard(false);
        synchronized (namingFuzzyWatchContext) {
            FuzzyWatchEventWatcherWrapper fuzzyWatchEventWatcherWrapper = new FuzzyWatchEventWatcherWrapper(watcher);
            if (namingFuzzyWatchContext.getFuzzyWatchEventWatcherWrappers().add(fuzzyWatchEventWatcherWrapper)) {
                LOGGER.info(" [add-watcher-ok] groupKeyPattern={}, watcher={},uuid={} ", groupKeyPattern, watcher,
                        fuzzyWatchEventWatcherWrapper.getUuid());
                Set<String> receivedServiceKeys = namingFuzzyWatchContext.getReceivedServiceKeys();
                if (CollectionUtils.isNotEmpty(receivedServiceKeys)) {
                    for (String serviceKey : receivedServiceKeys) {
                        NamingFuzzyWatchNotifyEvent namingFuzzyWatchNotifyEvent = NamingFuzzyWatchNotifyEvent.build(
                                notifierEventScope, groupKeyPattern, serviceKey, ADD_SERVICE, FUZZY_WATCH_INIT_NOTIFY,
                                fuzzyWatchEventWatcherWrapper.getUuid());
                        NotifyCenter.publishEvent(namingFuzzyWatchNotifyEvent);
                    }
                }
            }
        }
        
        return namingFuzzyWatchContext;
    }
    
    /**
     * init fuzzy watch context.
     *
     * @param groupKeyPattern groupKeyPattern.
     * @return fuzzy context.
     */
    public NamingFuzzyWatchContext initFuzzyWatchContextIfNeed(String groupKeyPattern) {
        if (!fuzzyMatchContextMap.containsKey(groupKeyPattern)) {
            synchronized (fuzzyMatchContextMap) {
                if (fuzzyMatchContextMap.containsKey(groupKeyPattern)) {
                    return fuzzyMatchContextMap.get(groupKeyPattern);
                }
                LOGGER.info("[fuzzy-watch] init fuzzy watch context for pattern {}", groupKeyPattern);
                fuzzyMatchContextMap.putIfAbsent(groupKeyPattern,
                        new NamingFuzzyWatchContext(notifierEventScope, groupKeyPattern));
                notifyFuzzyWatchSync();
            }
        }
        return fuzzyMatchContextMap.get(groupKeyPattern);
    }
    
    /**
     * remove fuzzy watch context for pattern.
     *
     * @param groupKeyPattern group key pattern.
     */
    public synchronized void removePatternMatchCache(String groupKeyPattern) {
        NamingFuzzyWatchContext namingFuzzyWatchContext = fuzzyMatchContextMap.get(groupKeyPattern);
        if (namingFuzzyWatchContext == null) {
            return;
        }
        if (namingFuzzyWatchContext.isDiscard() && namingFuzzyWatchContext.getFuzzyWatchEventWatcherWrappers()
                .isEmpty()) {
            LOGGER.info("[fuzzy-watch] remove fuzzy watch context for pattern {}", groupKeyPattern);
            fuzzyMatchContextMap.remove(groupKeyPattern);
        }
    }
    
    /**
     * notify sync fuzzy watch with server.
     */
    void notifyFuzzyWatchSync() {
        fuzzyWatchExecuteBell.offer(bellItem);
    }
    
    /**
     * Execute fuzzy listen configuration changes.
     *
     * <p>This method iterates through all fuzzy listen contexts and determines whether they need to be added or
     * removed based on their consistency with the server and discard status. It then calls the appropriate method to
     * execute the fuzzy listen operation.
     *
     * @throws NacosException If an error occurs during the execution of fuzzy listen configuration changes.
     */
    public void executeNamingFuzzyWatch() throws NacosException {
        
        // Obtain the current timestamp
        long now = System.currentTimeMillis();
        
        // Determine whether a full synchronization is needed
        boolean needAllSync = now - fuzzyWatchLastAllSyncTime.get() >= FUZZY_LISTEN_ALL_SYNC_INTERNAL;
        
        List<NamingFuzzyWatchContext> needSyncContexts = new ArrayList<>();
        // Iterate through all fuzzy listen contexts
        for (NamingFuzzyWatchContext context : fuzzyMatchContextMap.values()) {
            // Check if the context is consistent with the server
            if (context.isConsistentWithServer()) {
                context.syncFuzzyWatchers();
                // Skip if a full synchronization is not needed
                if (!needAllSync) {
                    continue;
                }
            }
            
            needSyncContexts.add(context);
        }
        
        // Execute fuzzy listen operation for addition
        doExecuteNamingFuzzyWatch(needSyncContexts);
        
        // Update last all sync time if a full synchronization was performed
        if (needAllSync) {
            fuzzyWatchLastAllSyncTime.set(now);
        }
    }
    
    public void resetConsistenceStatus() {
        fuzzyMatchContextMap.values()
                .forEach(fuzzyWatcherContext -> fuzzyWatcherContext.setConsistentWithServer(false));
    }
    
    /**
     * Execute fuzzy listen configuration changes for a specific map of contexts.
     *
     * <p>This method submits tasks to execute fuzzy listen operations asynchronously for the provided contexts. It
     * waits for all tasks to complete and logs any errors that occur.
     *
     * @param contextLists The map of contexts to execute fuzzy listen operations for.
     * @throws NacosException If an error occurs during the execution of fuzzy listen configuration changes.
     */
    private void doExecuteNamingFuzzyWatch(List<NamingFuzzyWatchContext> contextLists) throws NacosException {
        // Return if the context map is null or empty
        if (CollectionUtils.isEmpty(contextLists)) {
            return;
        }
        
        // Iterate through the context map and submit tasks for execution
        for (NamingFuzzyWatchContext entry : contextLists) {
            // Submit task for execution
            NamingFuzzyWatchRequest configFuzzyWatchRequest = buildFuzzyWatchNamingRequest(entry);
            try {
                
                // Execute the fuzzy listen operation
                NamingFuzzyWatchResponse listenResponse = namingGrpcClientProxy.fuzzyWatchRequest(
                        configFuzzyWatchRequest);
                if (listenResponse != null && listenResponse.isSuccess()) {
                    
                    if (configFuzzyWatchRequest.getWatchType().equals(WATCH_TYPE_CANCEL_WATCH)) {
                        removePatternMatchCache(entry.getGroupKeyPattern());
                    } else {
                        entry.setConsistentWithServer(true);
                    }
                    entry.clearOverLimitTs();
                }
                
            } catch (NacosException e) {
                // Log error and retry after a short delay
                
                if (FUZZY_WATCH_PATTERN_OVER_LIMIT.getCode() == e.getErrCode()
                        || FUZZY_WATCH_PATTERN_MATCH_COUNT_OVER_LIMIT.getCode() == e.getErrCode()) {
                    LOGGER.error(" fuzzy watch pattern over limit,pattern ->{} ,fuzzy watch will be suppressed,msg={}",
                            entry.getGroupKeyPattern(), e.getErrMsg());
                    NamingFuzzyWatchLoadEvent namingFuzzyWatchLoadEvent = NamingFuzzyWatchLoadEvent.buildEvent(
                            e.getErrCode(), entry.getGroupKeyPattern(), notifierEventScope);
                    NotifyCenter.publishEvent(namingFuzzyWatchLoadEvent);
                    
                } else {
                    LOGGER.error(" fuzzy watch request fail.", e);
                    
                    try {
                        Thread.sleep(1000L);
                    } catch (InterruptedException interruptedException) {
                        // Ignore interruption
                    }
                    // Retry notification
                    notifyFuzzyWatchSync();
                }
                
            }
        }
        
    }
    
    private NamingFuzzyWatchRequest buildFuzzyWatchNamingRequest(NamingFuzzyWatchContext namingFuzzyWatchContext) {
        NamingFuzzyWatchRequest namingFuzzyWatchRequest = new NamingFuzzyWatchRequest();
        namingFuzzyWatchRequest.setInitializing(namingFuzzyWatchContext.isInitializing());
        namingFuzzyWatchRequest.setNamespace(namingGrpcClientProxy.getNamespaceId());
        namingFuzzyWatchRequest.setReceivedGroupKeys(namingFuzzyWatchContext.getReceivedServiceKeys());
        namingFuzzyWatchRequest.setGroupKeyPattern(namingFuzzyWatchContext.getGroupKeyPattern());
        if (namingFuzzyWatchContext.isDiscard() && namingFuzzyWatchContext.getFuzzyWatchEventWatcherWrappers()
                .isEmpty()) {
            namingFuzzyWatchRequest.setWatchType(WATCH_TYPE_CANCEL_WATCH);
        } else {
            namingFuzzyWatchRequest.setWatchType(WATCH_TYPE_WATCH);
        }
        return namingFuzzyWatchRequest;
    }
    
    public Map<String, NamingFuzzyWatchContext> getFuzzyMatchContextMap() {
        return fuzzyMatchContextMap;
    }
    
    @Override
    public void onEvent(Event event) {
        
        if (event instanceof NamingFuzzyWatchNotifyEvent) {
            if (!event.scope().equals(notifierEventScope)) {
                return;
            }
            NamingFuzzyWatchNotifyEvent watchNotifyEvent = (NamingFuzzyWatchNotifyEvent) event;
            String changedType = watchNotifyEvent.getChangedType();
            String syncType = watchNotifyEvent.getSyncType();
            
            String serviceKey = watchNotifyEvent.getServiceKey();
            String pattern = watchNotifyEvent.getPattern();
            String watchUuid = watchNotifyEvent.getWatcherUuid();
            NamingFuzzyWatchContext namingFuzzyWatchContext = fuzzyMatchContextMap.get(pattern);
            if (namingFuzzyWatchContext == null) {
                return;
            }
            namingFuzzyWatchContext.notifyFuzzyWatchers(serviceKey, changedType, syncType, watchUuid);
        }
        if (event instanceof NamingFuzzyWatchLoadEvent) {
            if (!event.scope().equals(notifierEventScope)) {
                return;
            }
            
            NamingFuzzyWatchLoadEvent overLimitEvent = (NamingFuzzyWatchLoadEvent) event;
            NamingFuzzyWatchContext namingFuzzyWatchContext = fuzzyMatchContextMap.get(
                    overLimitEvent.getGroupKeyPattern());
            if (namingFuzzyWatchContext == null) {
                return;
            }
            
            namingFuzzyWatchContext.notifyOverLimitWatchers(overLimitEvent.getCode());
        }
        
    }
    
    @Override
    public List<Class<? extends Event>> subscribeTypes() {
        List<Class<? extends Event>> result = new LinkedList<>();
        result.add(NamingFuzzyWatchNotifyEvent.class);
        result.add(NamingFuzzyWatchLoadEvent.class);
        return result;
    }
    
    public String getNotifierEventScope() {
        return notifierEventScope;
    }
}
