/*
 *  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.distributed.raft;

import com.alibaba.nacos.common.model.RestResult;
import com.alibaba.nacos.common.model.RestResultUtils;
import com.alibaba.nacos.consistency.RequestProcessor;
import com.alibaba.nacos.consistency.cp.RequestProcessor4CP;
import com.alibaba.nacos.consistency.entity.ReadRequest;
import com.alibaba.nacos.consistency.entity.Response;
import com.alibaba.nacos.consistency.entity.WriteRequest;
import com.alibaba.nacos.core.cluster.Member;
import com.alibaba.nacos.core.distributed.ProtocolManager;
import com.alibaba.nacos.core.distributed.raft.utils.FailoverClosure;
import com.alibaba.nacos.sys.env.EnvUtil;
import com.alipay.sofa.jraft.CliService;
import com.alipay.sofa.jraft.Node;
import com.alipay.sofa.jraft.RaftGroupService;
import com.alipay.sofa.jraft.RouteTable;
import com.alipay.sofa.jraft.Status;
import com.alipay.sofa.jraft.conf.Configuration;
import com.alipay.sofa.jraft.core.NodeImpl;
import com.alipay.sofa.jraft.core.State;
import com.alipay.sofa.jraft.entity.PeerId;
import com.alipay.sofa.jraft.error.RemotingException;
import com.alipay.sofa.jraft.rpc.CliRequests;
import com.alipay.sofa.jraft.rpc.InvokeCallback;
import com.alipay.sofa.jraft.rpc.RpcClient;
import com.alipay.sofa.jraft.rpc.impl.FutureImpl;
import com.alipay.sofa.jraft.rpc.impl.cli.CliClientServiceImpl;
import com.alipay.sofa.jraft.util.Endpoint;
import com.google.protobuf.Message;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
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.junit.jupiter.MockitoExtension;
import org.mockito.junit.jupiter.MockitoSettings;
import org.mockito.quality.Strictness;
import org.springframework.mock.env.MockEnvironment;
import org.springframework.test.util.ReflectionTestUtils;

import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;

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.eq;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

@ExtendWith(MockitoExtension.class)
// todo remove this
@MockitoSettings(strictness = Strictness.LENIENT)
class JRaftServerTest {
    
    private final String groupId = "test_group";
    
    private PeerId peerId1;
    
    private PeerId peerId2;
    
    private PeerId peerId3;
    
    private Configuration conf;
    
    private JRaftServer server;
    
    private RaftConfig config;
    
    @Mock
    private RequestProcessor4CP mockProcessor4CP;
    
    @Mock
    private CliClientServiceImpl cliClientServiceMock;
    
    @Mock
    private CliService cliServiceMock;
    
    @Mock
    private RpcClient rpcClient;
    
    @Mock
    private CompletableFuture<Response> future;
    
    private ReadRequest readRequest;
    
    @Mock
    private Node node;
    
    @Mock
    private RequestProcessor requestProcessor;
    
    @Mock
    private RaftGroupService raftGroupService;
    
    @Mock
    private NacosStateMachine nacosStateMachine;
    
    @BeforeAll
    static void beforeClass() {
        EnvUtil.setEnvironment(new MockEnvironment());
    }
    
    @BeforeEach
    void before() throws NoSuchFieldException, IllegalAccessException {
        initPeersAndConfiguration();
        config = new RaftConfig();
        Collection<Member> initEvent = Collections.singletonList(Member.builder().ip("1.1.1.1").port(7848).build());
        config.setMembers("1.1.1.1:7848", ProtocolManager.toCPMembersInfo(initEvent));
        
        server = new JRaftServer() {
            
            @Override
            boolean peerChange(JRaftMaintainService maintainService, Set<String> newPeers) {
                return super.peerChange(maintainService, newPeers);
            }
        };
        
        server.init(config);
        
        Map<String, JRaftServer.RaftGroupTuple> map = new HashMap<>();
        map.put("test_nacos",
                new JRaftServer.RaftGroupTuple(node, requestProcessor, raftGroupService, nacosStateMachine));
        server.mockMultiRaftGroup(map);
        
        mockcliClientService();
        mockcliService();
        setLeaderAs(peerId1);
        // Inject the mocked cliClientServiceMock into server.
        Field cliClientServiceField = JRaftServer.class.getDeclaredField("cliClientService");
        cliClientServiceField.setAccessible(true);
        cliClientServiceField.set(server, cliClientServiceMock);
        
        // Inject the mocked cliServiceMock into server.
        Field cliServiceField = JRaftServer.class.getDeclaredField("cliService");
        cliServiceField.setAccessible(true);
        cliServiceField.set(server, cliServiceMock);
        
        // currently useless
        ReadRequest.Builder readRequestBuilder = ReadRequest.newBuilder();
        readRequest = readRequestBuilder.build();
        
        when(mockProcessor4CP.loadSnapshotOperate()).thenReturn(Collections.emptyList());
        when(mockProcessor4CP.group()).thenReturn(groupId);
        
        when(future.completeExceptionally(any(IllegalArgumentException.class))).thenReturn(true);
        
        Field isStartedField = JRaftServer.class.getDeclaredField("isStarted");
        isStartedField.setAccessible(true);
        isStartedField.set(server, true);
    }
    
    private void initPeersAndConfiguration() {
        peerId1 = new PeerId("11.11.11.11", 7848);
        peerId2 = new PeerId("22.22.22.22", 7848);
        peerId3 = new PeerId("33.33.33.33", 7848);
        this.conf = new Configuration();
        conf.addPeer(peerId1);
        conf.addPeer(peerId2);
        conf.addPeer(peerId3);
        RouteTable.getInstance().updateConfiguration(groupId, conf);
    }
    
    private void mockcliClientService() {
        when(cliClientServiceMock.connect(any(Endpoint.class))).thenReturn(true);
        // Assign PeerId1 as the leader
        final CliRequests.GetLeaderRequest.Builder rb = CliRequests.GetLeaderRequest.newBuilder();
        rb.setGroupId(groupId);
        rb.build();
        final FutureImpl<Message> getLeaderFuture = new FutureImpl<>();
        final CliRequests.GetLeaderResponse.Builder gb = CliRequests.GetLeaderResponse.newBuilder();
        gb.setLeaderId(peerId1.toString());
        final CliRequests.GetLeaderResponse getLeaderResponse = gb.build();
        getLeaderFuture.setResult(getLeaderResponse);
    }
    
    private void mockcliService() {
        List<PeerId> peerIds = new ArrayList<>();
        peerIds.add(peerId1);
        peerIds.add(peerId2);
        peerIds.add(peerId3);
        when(cliServiceMock.getPeers(groupId, conf)).thenReturn(peerIds);
        when(cliServiceMock.addPeer(eq(groupId), eq(conf), any(PeerId.class))).thenReturn(Status.OK());
    }
    
    private void setLeaderAs(PeerId peerId) {
        RouteTable.getInstance().updateLeader(groupId, peerId);
    }
    
    @Test
    void testInvokeToLeader()
            throws NoSuchMethodException, InvocationTargetException, IllegalAccessException, RemotingException, InterruptedException {
        when(cliClientServiceMock.getRpcClient()).thenReturn(rpcClient);
        setLeaderAs(peerId1);
        int timeout = 3000;
        Method invokeToLeaderMethod = JRaftServer.class.getDeclaredMethod("invokeToLeader", String.class, Message.class,
                int.class, FailoverClosure.class);
        invokeToLeaderMethod.setAccessible(true);
        invokeToLeaderMethod.invoke(server, groupId, this.readRequest, timeout, null);
        verify(cliClientServiceMock).getRpcClient();
        verify(rpcClient).invokeAsync(eq(peerId1.getEndpoint()), eq(readRequest), any(InvokeCallback.class),
                any(long.class));
    }
    
    @Test
    void testRefreshRouteTable() {
        server.refreshRouteTable(groupId);
        verify(cliClientServiceMock, times(1)).connect(peerId1.getEndpoint());
        verify(cliClientServiceMock).getLeader(eq(peerId1.getEndpoint()), any(CliRequests.GetLeaderRequest.class),
                eq(null));
    }
    
    @Test
    void testCommit() throws NoSuchFieldException, IllegalAccessException, TimeoutException, InterruptedException {
        WriteRequest.Builder writeRequestBuilder = WriteRequest.newBuilder();
        WriteRequest writeRequest = writeRequestBuilder.build();
        
        // No group is set, and make sure that an IllegalArgumentException will be thrown.
        CompletableFuture<Response> future = server.commit(groupId, writeRequest, this.future);
        verify(future).completeExceptionally(any(IllegalArgumentException.class));
        
        // Set an group.
        Collection<RequestProcessor4CP> processors = Collections.singletonList(mockProcessor4CP);
        server.createMultiRaftGroup(processors);
        
        Field cliClientServiceField = JRaftServer.class.getDeclaredField("cliClientService");
        cliClientServiceField.setAccessible(true);
        cliClientServiceField.set(server, cliClientServiceMock);
        
        // Make the node leader and verify the invokeToLeader is never called.
        NodeImpl node = (NodeImpl) server.findNodeByGroup(groupId);
        Field stateField = NodeImpl.class.getDeclaredField("state");
        stateField.setAccessible(true);
        stateField.set(node, State.STATE_LEADER);
        
        server.commit(groupId, writeRequest, future);
        verify(cliClientServiceMock, never()).getRpcClient();
        
        // make the node follower and verify the invokeToLeader is called.
        node = (NodeImpl) server.findNodeByGroup(groupId);
        stateField.setAccessible(true);
        stateField.set(node, State.STATE_FOLLOWER);
        
        RouteTable.getInstance().updateLeader(groupId, peerId1);
        server.commit(groupId, writeRequest, future);
        verify(cliClientServiceMock).getRpcClient();
    }
    
    @Test
    void testRegisterSelfToCluster() {
        PeerId selfPeerId = new PeerId("4.4.4.4", 8080);
        server.registerSelfToCluster(groupId, selfPeerId, conf);
        verify(cliServiceMock).addPeer(groupId, conf, selfPeerId);
    }
    
    @Test
    void testPeerChange() {
        AtomicBoolean changed = new AtomicBoolean(false);
        
        JRaftMaintainService service = new JRaftMaintainService(server) {
            @Override
            public RestResult<String> execute(Map<String, String> args) {
                changed.set(true);
                return RestResultUtils.success();
            }
        };
        
        Collection<Member> firstEvent = Arrays.asList(Member.builder().ip("1.1.1.1").port(7848).build(),
                Member.builder().ip("127.0.0.1").port(80).build(), Member.builder().ip("127.0.0.2").port(81).build(),
                Member.builder().ip("127.0.0.3").port(82).build());
        server.peerChange(service, ProtocolManager.toCPMembersInfo(firstEvent));
        assertFalse(changed.get());
        changed.set(false);
        
        Collection<Member> secondEvent = Arrays.asList(Member.builder().ip("1.1.1.1").port(7848).build(),
                Member.builder().ip("127.0.0.1").port(80).build(), Member.builder().ip("127.0.0.2").port(81).build(),
                Member.builder().ip("127.0.0.4").port(83).build());
        server.peerChange(service, ProtocolManager.toCPMembersInfo(secondEvent));
        assertTrue(changed.get());
        changed.set(false);
        
        Collection<Member> thirdEvent = Arrays.asList(Member.builder().ip("1.1.1.1").port(7848).build(),
                Member.builder().ip("127.0.0.2").port(81).build(), Member.builder().ip("127.0.0.5").port(82).build());
        server.peerChange(service, ProtocolManager.toCPMembersInfo(thirdEvent));
        assertTrue(changed.get());
        changed.set(false);
        
        Collection<Member> fourEvent = Arrays.asList(Member.builder().ip("1.1.1.1").port(7848).build(),
                Member.builder().ip("127.0.0.1").port(80).build());
        server.peerChange(service, ProtocolManager.toCPMembersInfo(fourEvent));
        assertTrue(changed.get());
        changed.set(false);
        
        Collection<Member> fiveEvent = Arrays.asList(Member.builder().ip("1.1.1.1").port(7848).build(),
                Member.builder().ip("127.0.0.1").port(80).build(), Member.builder().ip("127.0.0.3").port(81).build());
        server.peerChange(service, ProtocolManager.toCPMembersInfo(fiveEvent));
        assertFalse(changed.get());
        changed.set(false);
    }
    
    @Test
    void testIsReady() {
        assertTrue(server.isReady());
        config.setStrictMode(true);
        ((Collection<RequestProcessor4CP>) ReflectionTestUtils.getField(server, "processors")).add(mockProcessor4CP);
        assertTrue(server.isReady());
        setLeaderAs(null);
        assertFalse(server.isReady());
    }
    
    @AfterEach
    void shutdown() {
        server.shutdown();
    }
    
}
