/*
 * 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.core.cluster;

import com.alibaba.nacos.api.ability.ServerAbilities;
import com.alibaba.nacos.api.common.NodeState;
import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.remote.response.ResponseCode;
import com.alibaba.nacos.auth.config.NacosAuthConfigHolder;
import com.alibaba.nacos.auth.util.AuthHeaderUtil;
import com.alibaba.nacos.common.JustForTest;
import com.alibaba.nacos.common.http.Callback;
import com.alibaba.nacos.common.http.HttpClientBeanHolder;
import com.alibaba.nacos.common.http.HttpUtils;
import com.alibaba.nacos.common.http.client.NacosAsyncRestTemplate;
import com.alibaba.nacos.common.http.param.Header;
import com.alibaba.nacos.common.http.param.Query;
import com.alibaba.nacos.common.model.RestResult;
import com.alibaba.nacos.common.notify.Event;
import com.alibaba.nacos.common.notify.NotifyCenter;
import com.alibaba.nacos.common.notify.listener.Subscriber;
import com.alibaba.nacos.common.utils.ConcurrentHashSet;
import com.alibaba.nacos.common.utils.ExceptionUtil;
import com.alibaba.nacos.common.utils.JacksonUtils;
import com.alibaba.nacos.common.utils.StringUtils;
import com.alibaba.nacos.common.utils.VersionUtils;
import com.alibaba.nacos.core.ability.ServerAbilityInitializer;
import com.alibaba.nacos.core.ability.ServerAbilityInitializerHolder;
import com.alibaba.nacos.core.ability.control.ServerAbilityControlManager;
import com.alibaba.nacos.core.auth.NacosServerAuthConfig;
import com.alibaba.nacos.core.cluster.lookup.LookupFactory;
import com.alibaba.nacos.core.cluster.remote.ClusterRpcClientProxy;
import com.alibaba.nacos.core.cluster.remote.request.MemberReportRequest;
import com.alibaba.nacos.core.cluster.remote.response.MemberReportResponse;
import com.alibaba.nacos.core.utils.Commons;
import com.alibaba.nacos.core.utils.GenericType;
import com.alibaba.nacos.core.utils.GlobalExecutor;
import com.alibaba.nacos.core.utils.Loggers;
import com.alibaba.nacos.sys.env.Constants;
import com.alibaba.nacos.sys.env.EnvUtil;
import com.alibaba.nacos.sys.utils.ApplicationUtils;
import com.alibaba.nacos.sys.utils.InetUtils;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentSkipListMap;

import static com.alibaba.nacos.api.exception.NacosException.CLIENT_INVALID_PARAM;

/**
 * Cluster node management in Nacos.
 *
 * <p>{@link ServerMemberManager#init()} Cluster node manager initialization {@link ServerMemberManager#shutdown()} The
 * cluster node manager is down {@link ServerMemberManager#getSelf()} Gets local node information
 * {@link ServerMemberManager#getServerList()} Gets the cluster node dictionary
 * {@link ServerMemberManager#getMemberAddressInfos()} Gets the address information of the healthy member node
 * {@link ServerMemberManager#allMembers()} Gets a list of member information objects
 * {@link ServerMemberManager#allMembersWithoutSelf()} Gets a list of cluster member nodes with the exception of this
 * node {@link ServerMemberManager#hasMember(String)} Is there a node
 * {@link ServerMemberManager#memberChange(Collection)} The final node list changes the method, making the full size
 * more {@link ServerMemberManager#memberJoin(Collection)} Node join, can automatically trigger
 * {@link ServerMemberManager#memberLeave(Collection)} When the node leaves, only the interface call can be manually
 * triggered {@link ServerMemberManager#update(Member)} Update the target node information
 * {@link ServerMemberManager#isUnHealth(String)} Whether the target node is healthy
 * {@link ServerMemberManager#initAndStartLookup()} Initializes the addressing mode
 *
 * @author <a href="mailto:liaochuntao@live.com">liaochuntao</a>
 */
@Component(value = "serverMemberManager")
public class ServerMemberManager implements NacosMemberManager {
    
    private final NacosAsyncRestTemplate asyncRestTemplate = HttpClientBeanHolder.getNacosAsyncRestTemplate(
            Loggers.CORE);
    
    private static final int DEFAULT_SERVER_PORT = 8848;
    
    private static final String SERVER_PORT_PROPERTY = "nacos.server.main.port";
    
    private static final String MEMBER_CHANGE_EVENT_QUEUE_SIZE_PROPERTY = "nacos.member-change-event.queue.size";
    
    private static final int DEFAULT_MEMBER_CHANGE_EVENT_QUEUE_SIZE = 128;
    
    private static final long DEFAULT_TASK_DELAY_TIME = 5_000L;
    
    /**
     * Cluster node list.
     */
    private volatile ConcurrentSkipListMap<String, Member> serverList;
    
    /**
     * Is this node in the cluster list.
     */
    private static volatile boolean isInIpList = true;
    
    /**
     * port.
     */
    private int port;
    
    /**
     * Address information for the local node.
     */
    private String localAddress;
    
    /**
     * Addressing pattern instances.
     */
    private MemberLookup lookup;
    
    /**
     * self member obj.
     */
    private volatile Member self;
    
    private volatile long memberReportTs = System.currentTimeMillis();
    
    /**
     * here is always the node information of the "UP" state.
     */
    private volatile Set<String> memberAddressInfos = new ConcurrentHashSet<>();
    
    /**
     * Broadcast this node element information task.
     */
    private final MemberInfoReportTask infoReportTask = new MemberInfoReportTask();
    
    private final UnhealthyMemberInfoReportTask unhealthyMemberInfoReportTask = new UnhealthyMemberInfoReportTask();
    
    public ServerMemberManager() throws Exception {
        this.serverList = new ConcurrentSkipListMap<>();
        init();
    }
    
    protected void init() throws NacosException {
        Loggers.CORE.info("Nacos-related cluster resource initialization");
        this.port = EnvUtil.getProperty(SERVER_PORT_PROPERTY, Integer.class, DEFAULT_SERVER_PORT);
        this.localAddress = InetUtils.getSelfIP() + ":" + port;
        this.self = MemberUtil.singleParse(this.localAddress);
        this.self.setExtendVal(MemberMetaDataConstants.VERSION, VersionUtils.version);
        //works  for gray model upgrade,can delete after compatibility period.
        this.self.setExtendVal(MemberMetaDataConstants.SUPPORT_GRAY_MODEL, true);
        this.self.setGrpcReportEnabled(true);
        
        // init abilities.
        this.self.setAbilities(initMemberAbilities());
        
        serverList.put(self.getAddress(), self);
        
        // register NodeChangeEvent publisher to NotifyManager
        registerClusterEvent();
        
        // Initializes the lookup mode
        initAndStartLookup();
        
        if (serverList.isEmpty()) {
            throw new NacosException(NacosException.SERVER_ERROR, "cannot get serverlist, so exit.");
        }
        
        Loggers.CORE.info("The cluster resource is initialized");
    }
    
    /**
     * Init the ability of current node.
     *
     * @return ServerAbilities
     * @deprecated ability of current node and event cluster can be managed by {@link ServerAbilityControlManager}
     */
    private ServerAbilities initMemberAbilities() {
        ServerAbilities serverAbilities = new ServerAbilities();
        for (ServerAbilityInitializer each : ServerAbilityInitializerHolder.getInstance().getInitializers()) {
            each.initialize(serverAbilities);
        }
        return serverAbilities;
    }
    
    private void registerClusterEvent() {
        // Register node change events
        NotifyCenter.registerToPublisher(MembersChangeEvent.class,
                EnvUtil.getProperty(MEMBER_CHANGE_EVENT_QUEUE_SIZE_PROPERTY, Integer.class,
                        DEFAULT_MEMBER_CHANGE_EVENT_QUEUE_SIZE));
        
        // The address information of this node needs to be dynamically modified
        // when registering the IP change of this node
        NotifyCenter.registerSubscriber(new Subscriber<InetUtils.IPChangeEvent>() {
            @Override
            public void onEvent(InetUtils.IPChangeEvent event) {
                String newAddress = event.getNewIP() + ":" + port;
                ServerMemberManager.this.localAddress = newAddress;
                EnvUtil.setLocalAddress(localAddress);
                
                Member self = ServerMemberManager.this.self;
                self.setIp(event.getNewIP());
                
                String oldAddress = event.getOldIP() + ":" + port;
                ServerMemberManager.this.serverList.remove(oldAddress);
                ServerMemberManager.this.serverList.put(newAddress, self);
                
                ServerMemberManager.this.memberAddressInfos.remove(oldAddress);
                ServerMemberManager.this.memberAddressInfos.add(newAddress);
            }
            
            @Override
            public Class<? extends Event> subscribeType() {
                return InetUtils.IPChangeEvent.class;
            }
        });
    }
    
    private void initAndStartLookup() throws NacosException {
        this.lookup = LookupFactory.createLookUp(this);
        this.lookup.useAddressServer();
        this.lookup.start();
    }
    
    /**
     * switch look up.
     *
     * @param name look up name.
     * @throws NacosException exception.
     */
    public void switchLookup(String name) throws NacosException {
        this.lookup = LookupFactory.switchLookup(name, this);
        this.lookup.useAddressServer();
        this.lookup.start();
    }
    
    /**
     * member information update.
     *
     * @param newMember {@link Member}
     * @return update is success
     */
    public boolean update(Member newMember) {
        Loggers.CLUSTER.debug("member information update : {}", newMember);
        
        String address = newMember.getAddress();
        if (!serverList.containsKey(address)) {
            Loggers.CLUSTER.warn("address {} want to update Member, but not in member list!", newMember.getAddress());
            return false;
        }
        
        serverList.computeIfPresent(address, (s, member) -> {
            if (NodeState.DOWN.equals(newMember.getState())) {
                memberAddressInfos.remove(newMember.getAddress());
            }
            boolean isPublishChangeEvent = MemberUtil.isBasicInfoChanged(newMember, member);
            newMember.setExtendVal(MemberMetaDataConstants.LAST_REFRESH_TIME, System.currentTimeMillis());
            MemberUtil.copy(newMember, member);
            if (isPublishChangeEvent) {
                // member basic data changes and all listeners need to be notified
                notifyMemberChange(member);
            }
            return member;
        });
        
        return true;
    }
    
    void notifyMemberChange(Member member) {
        NotifyCenter.publishEvent(MembersChangeEvent.builder().trigger(member).members(allMembers()).build());
    }
    
    /**
     * Whether the node exists within the cluster.
     *
     * @param address ip:port
     * @return is exists
     */
    public boolean hasMember(String address) {
        boolean result = serverList.containsKey(address);
        if (result) {
            return true;
        }
        
        // If only IP information is passed in, a fuzzy match is required
        for (Map.Entry<String, Member> entry : serverList.entrySet()) {
            if (StringUtils.contains(entry.getKey(), address)) {
                result = true;
                break;
            }
        }
        return result;
    }
    
    public List<String> getServerListUnhealth() {
        List<String> unhealthyMembers = new ArrayList<>();
        for (Member member : this.allMembers()) {
            NodeState state = member.getState();
            if (state.equals(NodeState.DOWN)) {
                unhealthyMembers.add(member.getAddress());
            }
            
        }
        return unhealthyMembers;
    }
    
    public MemberLookup getLookup() {
        return lookup;
    }
    
    public Member getSelf() {
        return this.self;
    }
    
    public Member find(String address) {
        return serverList.get(address);
    }
    
    /**
     * return this cluster all members.
     *
     * @return {@link Collection} all member
     */
    @Override
    public Collection<Member> allMembers() {
        // We need to do a copy to avoid affecting the real data
        HashSet<Member> set = new HashSet<>(serverList.values());
        set.add(self);
        return set;
    }
    
    /**
     * return this cluster all members without self.
     *
     * @return {@link Collection} all member without self
     */
    public List<Member> allMembersWithoutSelf() {
        List<Member> members = new ArrayList<>(serverList.values());
        members.remove(self);
        return members;
    }
    
    @Override
    public synchronized boolean memberChange(Collection<Member> members) {
        
        if (members == null || members.isEmpty()) {
            return false;
        }
        
        boolean isContainSelfIp = members.stream()
                .anyMatch(ipPortTmp -> Objects.equals(localAddress, ipPortTmp.getAddress()));
        
        if (isContainSelfIp) {
            isInIpList = true;
        } else {
            isInIpList = false;
            members.add(this.self);
            Loggers.CLUSTER.warn("[serverlist] self ip {} not in serverlist {}", self, members);
        }
        
        // If the number of old and new clusters is different, the cluster information
        // must have changed; if the number of clusters is the same, then compare whether
        // there is a difference; if there is a difference, then the cluster node changes
        // are involved and all recipients need to be notified of the node change event
        
        boolean hasChange = members.size() != serverList.size();
        ConcurrentSkipListMap<String, Member> tmpMap = new ConcurrentSkipListMap<>();
        Set<String> tmpAddressInfo = new ConcurrentHashSet<>();
        for (Member member : members) {
            final String address = member.getAddress();
            
            Member existMember = serverList.get(address);
            if (existMember == null) {
                hasChange = true;
                tmpMap.put(address, member);
            } else {
                //to keep extendInfo and abilities that report dynamically.
                tmpMap.put(address, existMember);
            }
            
            if (NodeState.UP.equals(member.getState())) {
                tmpAddressInfo.add(address);
            }
        }
        
        serverList = tmpMap;
        memberAddressInfos = tmpAddressInfo;
        
        Collection<Member> finalMembers = allMembers();
        
        // Persist the current cluster node information to cluster.conf
        // <important> need to put the event publication into a synchronized block to ensure
        // that the event publication is sequential
        if (hasChange) {
            Loggers.CLUSTER.info("[serverlist] changed to : {}", finalMembers);
            MemberUtil.syncToFile(finalMembers);
            Event event = MembersChangeEvent.builder().members(finalMembers).build();
            NotifyCenter.publishEvent(event);
        } else {
            if (Loggers.CLUSTER.isDebugEnabled()) {
                Loggers.CLUSTER.debug("[serverlist] not updated, is still : {}", finalMembers);
            }
        }
        
        return hasChange;
    }
    
    /**
     * members join this cluster.
     *
     * @param members {@link Collection} new members
     * @return is success
     */
    public synchronized boolean memberJoin(Collection<Member> members) {
        Set<Member> set = new HashSet<>(members);
        set.addAll(allMembers());
        return memberChange(set);
    }
    
    /**
     * members leave this cluster.
     *
     * @param members {@link Collection} wait leave members
     * @return is success
     */
    public synchronized boolean memberLeave(Collection<Member> members) {
        Set<Member> set = new HashSet<>(allMembers());
        set.removeAll(members);
        return memberChange(set);
    }
    
    /**
     * check this member whether is in the specific status.
     *
     * @param address ip:port
     * @return is health
     */
    public boolean stateCheck(String address, List<NodeState> nodeStates) {
        Member member = serverList.get(address);
        if (member == null) {
            return false;
        }
        for (NodeState nodeState : nodeStates) {
            if (nodeState.equals(member.getState())) {
                return true;
            }
        }
        return false;
    }
    
    /**
     * this member {@link Member#getState()} is health.
     *
     * @param address ip:port
     * @return is health
     */
    public boolean isUnHealth(String address) {
        Member member = serverList.get(address);
        if (member == null) {
            return false;
        }
        return !NodeState.UP.equals(member.getState());
    }
    
    public boolean isFirstIp() {
        return Objects.equals(serverList.firstKey(), this.localAddress);
    }
    
    public void setSelfReady(int port) {
        getSelf().setState(NodeState.UP);
        if (!EnvUtil.getStandaloneMode()) {
            GlobalExecutor.scheduleByCommon(this.infoReportTask, DEFAULT_TASK_DELAY_TIME);
            GlobalExecutor.scheduleByCommon(this.unhealthyMemberInfoReportTask, DEFAULT_TASK_DELAY_TIME);
        }
        EnvUtil.setPort(port);
        EnvUtil.setLocalAddress(this.localAddress);
        Loggers.CLUSTER.info("This node is ready to provide external services");
    }
    
    /**
     * ServerMemberManager shutdown.
     *
     * @throws NacosException NacosException
     */
    @PreDestroy
    public void shutdown() throws NacosException {
        serverList.clear();
        memberAddressInfos.clear();
        infoReportTask.shutdown();
        LookupFactory.destroy();
    }
    
    public Set<String> getMemberAddressInfos() {
        return memberAddressInfos;
    }
    
    @JustForTest
    public void updateMember(Member member) {
        serverList.put(member.getAddress(), member);
    }
    
    @JustForTest
    public MemberInfoReportTask getInfoReportTask() {
        return infoReportTask;
    }
    
    public Map<String, Member> getServerList() {
        return Collections.unmodifiableMap(serverList);
    }
    
    public static boolean isInIpList() {
        return isInIpList;
    }
    
    // Synchronize the metadata information of a node
    // A health check of the target node is also attached
    
    class MemberInfoReportTask extends Task {
        
        private final GenericType<RestResult<String>> reference = new GenericType<RestResult<String>>() {
        };
        
        private int cursor = 0;
        
        private ClusterRpcClientProxy clusterRpcClientProxy;
        
        public static final long REPORT_INTERVAL = 50000L;
        
        @Override
        protected void executeBody() {
            List<Member> members = ServerMemberManager.this.allMembersWithoutSelf();
            
            //report member count per 50 seconds.
            if (System.currentTimeMillis() - memberReportTs > REPORT_INTERVAL) {
                Loggers.CLUSTER.info("[serverlist] membercount={}", members.size() + 1);
                memberReportTs = System.currentTimeMillis();
            }
            
            if (members.isEmpty()) {
                return;
            }
            
            this.cursor = (this.cursor + 1) % members.size();
            Member target = members.get(cursor);
            
            Loggers.CLUSTER.debug("report the metadata to the node : {}", target.getAddress());
            
            // adapt old version
            if (target.getAbilities().getRemoteAbility().isGrpcReportEnabled() || target.isGrpcReportEnabled()) {
                reportByGrpc(target);
            } else {
                reportByHttp(target);
            }
        }
        
        protected void reportByHttp(Member target) {
            final String url = HttpUtils.buildUrl(false, target.getAddress(), EnvUtil.getContextPath(),
                    Commons.NACOS_CORE_CONTEXT, "/cluster/report");
            
            try {
                Header header = Header.newInstance().addParam(Constants.NACOS_SERVER_HEADER, VersionUtils.version);
                AuthHeaderUtil.addIdentityToHeader(header, NacosAuthConfigHolder.getInstance()
                        .getNacosAuthConfigByScope(NacosServerAuthConfig.NACOS_SERVER_AUTH_SCOPE));
                asyncRestTemplate.post(url, header, Query.EMPTY, getSelf(), reference.getType(),
                        new Callback<String>() {
                            @Override
                            public void onReceive(RestResult<String> result) {
                                if (result.ok()) {
                                    handleReportResult(result.getData(), target);
                                } else {
                                    Loggers.CLUSTER.warn("failed to report new info to target node : {}, result : {}",
                                            target.getAddress(), result);
                                    MemberUtil.onFail(ServerMemberManager.this, target);
                                    // try to connect by grpc next time, adapt old version
                                    target.setGrpcReportEnabled(true);
                                    target.getAbilities().getRemoteAbility().setGrpcReportEnabled(true);
                                }
                            }
                            
                            @Override
                            public void onError(Throwable throwable) {
                                Loggers.CLUSTER.error("failed to report new info to target node : {}, error : {}",
                                        target.getAddress(), ExceptionUtil.getAllExceptionMsg(throwable));
                                MemberUtil.onFail(ServerMemberManager.this, target, throwable);
                                // try to connect by grpc next time, adapt old version
                                target.setGrpcReportEnabled(true);
                                target.getAbilities().getRemoteAbility().setGrpcReportEnabled(true);
                            }
                            
                            @Override
                            public void onCancel() {
                            
                            }
                        });
            } catch (Throwable ex) {
                Loggers.CLUSTER.error("failed to report new info to target node by http : {}, error : {}",
                        target.getAddress(), ExceptionUtil.getAllExceptionMsg(ex));
                // try to connect by grpc next time, adapt old version
                target.setGrpcReportEnabled(true);
                target.getAbilities().getRemoteAbility().setGrpcReportEnabled(true);
            }
        }
        
        protected void reportByGrpc(Member target) {
            //Todo  circular reference
            if (Objects.isNull(clusterRpcClientProxy)) {
                clusterRpcClientProxy = ApplicationUtils.getBean(ClusterRpcClientProxy.class);
            }
            if (!clusterRpcClientProxy.isRunning(target)) {
                MemberUtil.onFail(ServerMemberManager.this, target,
                        new NacosException(CLIENT_INVALID_PARAM, "No rpc client related to member: " + target));
                return;
            }
            
            MemberReportRequest memberReportRequest = new MemberReportRequest(getSelf());
            
            try {
                MemberReportResponse response = (MemberReportResponse) clusterRpcClientProxy.sendRequest(target,
                        memberReportRequest);
                if (response.getResultCode() == ResponseCode.SUCCESS.getCode()) {
                    MemberUtil.onSuccess(ServerMemberManager.this, target, response.getNode());
                } else {
                    MemberUtil.onFail(ServerMemberManager.this, target);
                }
            } catch (NacosException e) {
                if (e.getErrCode() == NacosException.NO_HANDLER) {
                    target.getAbilities().getRemoteAbility().setGrpcReportEnabled(false);
                    target.setGrpcReportEnabled(false);
                }
                Loggers.CLUSTER.error("failed to report new info to target node by grpc : {}, error : {}",
                        target.getAddress(), ExceptionUtil.getAllExceptionMsg(e));
            }
        }
        
        @Override
        protected void after() {
            GlobalExecutor.scheduleByCommon(this, 2_000L);
        }
        
        private void handleReportResult(String reportResult, Member target) {
            if (isBooleanResult(reportResult)) {
                MemberUtil.onSuccess(ServerMemberManager.this, target);
                return;
            }
            try {
                Member member = JacksonUtils.toObj(reportResult, Member.class);
                MemberUtil.onSuccess(ServerMemberManager.this, target, member);
            } catch (Exception e) {
                Loggers.CLUSTER.warn("Receive invalid report result from target {}, context {}", target.getAddress(),
                        reportResult);
                MemberUtil.onSuccess(ServerMemberManager.this, target);
            }
        }
        
        private boolean isBooleanResult(String reportResult) {
            return Boolean.TRUE.toString().equals(reportResult) || Boolean.FALSE.toString().equals(reportResult);
        }
    }
    
    class UnhealthyMemberInfoReportTask extends MemberInfoReportTask {
        
        @Override
        protected void executeBody() {
            List<Member> members = ServerMemberManager.this.allMembersWithoutSelf();
            
            if (members.isEmpty()) {
                return;
            }
            for (Member member : members) {
                if (!member.getState().equals(NodeState.UP)) {
                    if (member.getAbilities().getRemoteAbility().isGrpcReportEnabled()) {
                        reportByGrpc(member);
                    } else {
                        reportByHttp(member);
                    }
                    Loggers.CLUSTER.warn("report the metadata to the unhealthy node : {}", member.getAddress());
                    
                }
            }
        }
        
        @Override
        protected void after() {
            GlobalExecutor.scheduleByCommon(this, 5_000L);
        }
    }
    
}
