/*
 * 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.test.config;

import com.alibaba.nacos.api.config.listener.AbstractListener;
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.ThreadUtils;
import com.alibaba.nacos.config.server.model.event.RaftDbErrorRecoverEvent;
import com.alibaba.nacos.config.server.service.repository.ConfigInfoPersistService;
import com.alibaba.nacos.config.server.service.repository.embedded.EmbeddedConfigInfoPersistServiceImpl;
import com.alibaba.nacos.consistency.cp.CPProtocol;
import com.alibaba.nacos.core.distributed.id.IdGeneratorManager;
import com.alibaba.nacos.core.distributed.raft.utils.JRaftConstants;
import com.alibaba.nacos.core.utils.GenericType;
import com.alibaba.nacos.persistence.constants.PersistenceConstant;
import com.alibaba.nacos.persistence.model.event.RaftDbErrorEvent;
import com.alibaba.nacos.sys.utils.InetUtils;
import com.alibaba.nacos.test.base.BaseClusterTest;
import com.alibaba.nacos.test.base.ConfigCleanUtils;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.MethodOrderer.MethodName;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestMethodOrder;
import org.springframework.context.ConfigurableApplicationContext;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;

/**
 * Integration tests for Derby Raft configuration in the cluster environment.
 *
 * @author <a href="mailto:liaochuntao@live.com">liaochuntao</a>
 */
// todo the suffix is _DITCase, the case will be skipped by default
@SuppressWarnings("checkstyle:AbbreviationAsWordInName")
@TestMethodOrder(MethodName.class)
class ConfigDerbyRaftConfigITCase extends BaseClusterTest {
    
    @BeforeAll
    static void beforeClass() {
        ConfigCleanUtils.changeToNewTestNacosHome(ConfigDerbyRaftConfigITCase.class.getSimpleName());
    }
    
    @Test
    void testPublishConfigRaftCluster() throws Exception {
        boolean result = iconfig7.publishConfig("raft_test", "cluster_test_1", "this.is.raft_cluster=lessspring_7");
        assertTrue(result);
        
        ThreadUtils.sleep(5000);
        
        ConfigurableApplicationContext context7 = applications.get("8847");
        ConfigurableApplicationContext context8 = applications.get("8848");
        ConfigurableApplicationContext context9 = applications.get("8849");
        
        ConfigInfoPersistService operate7 = context7.getBean(EmbeddedConfigInfoPersistServiceImpl.class);
        ConfigInfoPersistService operate8 = context8.getBean(EmbeddedConfigInfoPersistServiceImpl.class);
        ConfigInfoPersistService operate9 = context9.getBean(EmbeddedConfigInfoPersistServiceImpl.class);
        
        String s7 = operate7.findConfigInfo("raft_test", "cluster_test_1", "").getContent();
        String s8 = operate8.findConfigInfo("raft_test", "cluster_test_1", "").getContent();
        String s9 = operate9.findConfigInfo("raft_test", "cluster_test_1", "").getContent();
        
        assertArrayEquals(new String[] {s7, s8, s9},
                new String[] {"this.is.raft_cluster=lessspring_7", "this.is.raft_cluster=lessspring_7",
                        "this.is.raft_cluster=lessspring_7"}, "The three nodes must have consistent data");
    }
    
    @Test
    void testPublishConfigCluster() throws Exception {
        ThreadUtils.sleep(5000);
        
        boolean result = iconfig8.publishConfig("raft_test", "cluster_test_2", "this.is.raft_cluster=lessspring_8");
        assertTrue(result);
        
        ThreadUtils.sleep(5000);
        
        ConfigurableApplicationContext context7 = applications.get("8847");
        ConfigurableApplicationContext context8 = applications.get("8848");
        ConfigurableApplicationContext context9 = applications.get("8849");
        
        ConfigInfoPersistService operate7 = context7.getBean(EmbeddedConfigInfoPersistServiceImpl.class);
        ConfigInfoPersistService operate8 = context8.getBean(EmbeddedConfigInfoPersistServiceImpl.class);
        ConfigInfoPersistService operate9 = context9.getBean(EmbeddedConfigInfoPersistServiceImpl.class);
        
        String s7 = operate7.findConfigInfo("raft_test", "cluster_test_2", "").getContent();
        String s8 = operate8.findConfigInfo("raft_test", "cluster_test_2", "").getContent();
        String s9 = operate9.findConfigInfo("raft_test", "cluster_test_2", "").getContent();
        
        assertArrayEquals(new String[] {s7, s8, s9},
                new String[] {"this.is.raft_cluster=lessspring_8", "this.is.raft_cluster=lessspring_8",
                        "this.is.raft_cluster=lessspring_8"}, "The three nodes must have consistent data");
    }
    
    @Test
    void testPublishConfigRaftClusterNew() throws Exception {
        ThreadUtils.sleep(5000);
        boolean result = iconfig9.publishConfig("raft_test", "cluster_test_2", "this.is.raft_cluster=lessspring_9");
        assertTrue(result);
        
        ThreadUtils.sleep(5000);
        
        ConfigurableApplicationContext context7 = applications.get("8847");
        ConfigurableApplicationContext context8 = applications.get("8848");
        ConfigurableApplicationContext context9 = applications.get("8849");
        
        ConfigInfoPersistService operate7 = context7.getBean(EmbeddedConfigInfoPersistServiceImpl.class);
        ConfigInfoPersistService operate8 = context8.getBean(EmbeddedConfigInfoPersistServiceImpl.class);
        ConfigInfoPersistService operate9 = context9.getBean(EmbeddedConfigInfoPersistServiceImpl.class);
        
        String s7 = operate7.findConfigInfo("raft_test", "cluster_test_2", "").getContent();
        String s8 = operate8.findConfigInfo("raft_test", "cluster_test_2", "").getContent();
        String s9 = operate9.findConfigInfo("raft_test", "cluster_test_2", "").getContent();
        
        assertArrayEquals(new String[] {s7, s8, s9},
                new String[] {"this.is.raft_cluster=lessspring_9", "this.is.raft_cluster=lessspring_9",
                        "this.is.raft_cluster=lessspring_9"}, "The three nodes must have consistent data");
    }
    
    @Test
    void testModifyConfigRaftCluster() throws Exception {
        boolean result = iconfig7.publishConfig("raft_test", "cluster_test_1",
                "this.is.raft_cluster=lessspring_7_it_is_for_modify");
        assertTrue(result);
        
        ThreadUtils.sleep(5000);
        
        ConfigurableApplicationContext context7 = applications.get("8847");
        ConfigurableApplicationContext context8 = applications.get("8848");
        ConfigurableApplicationContext context9 = applications.get("8849");
        
        ConfigInfoPersistService operate7 = context7.getBean(EmbeddedConfigInfoPersistServiceImpl.class);
        ConfigInfoPersistService operate8 = context8.getBean(EmbeddedConfigInfoPersistServiceImpl.class);
        ConfigInfoPersistService operate9 = context9.getBean(EmbeddedConfigInfoPersistServiceImpl.class);
        
        String s7 = operate7.findConfigInfo("raft_test", "cluster_test_1", "").getContent();
        String s8 = operate8.findConfigInfo("raft_test", "cluster_test_1", "").getContent();
        String s9 = operate9.findConfigInfo("raft_test", "cluster_test_1", "").getContent();
        
        assertArrayEquals(new String[] {s7, s8, s9}, new String[] {"this.is.raft_cluster=lessspring_7_it_is_for_modify",
                "this.is.raft_cluster=lessspring_7_it_is_for_modify",
                "this.is.raft_cluster=lessspring_7_it_is_for_modify"}, "The three nodes must have consistent data");
    }
    
    @Test
    void testClientOperationConfig() throws Exception {
        final String dataId = "test_l_client_operation";
        final String groupId = "test_l_client_operation";
        String content = "test_l_client_operation" + System.currentTimeMillis();
        
        // publish by 8847
        boolean result = iconfig7.publishConfig(dataId, groupId, content);
        assertTrue(result);
        ThreadUtils.sleep(5000);
        
        String v1Config7 = iconfig7.getConfig(dataId, groupId, 5000L);
        String v1Config8 = iconfig8.getConfig(dataId, groupId, 5000L);
        String v1Config9 = iconfig9.getConfig(dataId, groupId, 5000L);
        
        assertEquals(content, v1Config7);
        assertEquals(content, v1Config8);
        assertEquals(content, v1Config9);
        
        // publish by 8848
        content = "test_l_client_operation" + System.currentTimeMillis();
        result = iconfig8.publishConfig(dataId, groupId, content);
        assertTrue(result);
        ThreadUtils.sleep(5000);
        
        String v2Config7 = iconfig7.getConfig(dataId, groupId, 5000L);
        String v2Config8 = iconfig8.getConfig(dataId, groupId, 5000L);
        String v2Config9 = iconfig9.getConfig(dataId, groupId, 5000L);
        
        assertEquals(content, v2Config7);
        assertEquals(content, v2Config8);
        assertEquals(content, v2Config9);
        
        // publish by 8849
        content = "test_l_client_operation" + System.currentTimeMillis();
        result = iconfig9.publishConfig(dataId, groupId, content);
        assertTrue(result);
        ThreadUtils.sleep(5000);
        
        String v3Config7 = iconfig7.getConfig(dataId, groupId, 5000L);
        String v3Config8 = iconfig8.getConfig(dataId, groupId, 5000L);
        String v3Config9 = iconfig9.getConfig(dataId, groupId, 5000L);
        
        assertEquals(content, v3Config7);
        assertEquals(content, v3Config8);
        assertEquals(content, v3Config9);
        
        // delete by 8849
        result = iconfig9.removeConfig(dataId, groupId);
        assertTrue(result);
        ThreadUtils.sleep(5000);
        
        String v4Config7 = iconfig7.getConfig(dataId, groupId, 5000L);
        String v4Config8 = iconfig8.getConfig(dataId, groupId, 5000L);
        String v4Config9 = iconfig9.getConfig(dataId, groupId, 5000L);
        
        assertNull(v4Config7);
        assertNull(v4Config8);
        assertNull(v4Config9);
    }
    
    @Test
    void testConfigListener() throws Exception {
        String dataId = "test_h_config_listener";
        String group = "test_h_config_listener";
        String content = "test_h_config_listener";
        CountDownLatch[] latch = new CountDownLatch[] {new CountDownLatch(1), new CountDownLatch(0)};
        AtomicReference<String> r = new AtomicReference<>();
        AtomicInteger i = new AtomicInteger(0);
        iconfig7.addListener(dataId, group, new AbstractListener() {
            @Override
            public void receiveConfigInfo(String configInfo) {
                System.out.println(configInfo);
                r.set(configInfo);
                latch[i.getAndIncrement()].countDown();
            }
        });
        
        iconfig7.publishConfig(dataId, group, content);
        
        ThreadUtils.sleep(10_000L);
        latch[0].await(10_000L, TimeUnit.MILLISECONDS);
        assertEquals(content, r.get());
        assertEquals(content, iconfig7.getConfig(dataId, group, 2_000L));
        
        content = content + System.currentTimeMillis();
        iconfig7.publishConfig(dataId, group, content);
        
        ThreadUtils.sleep(10_000L);
        latch[1].await(10_000L, TimeUnit.MILLISECONDS);
        assertEquals(content, r.get());
        assertEquals(content, iconfig7.getConfig(dataId, group, 2_000L));
    }
    
    @Test
    void testDerbyOperations() throws Exception {
        String url = "http://127.0.0.1:8848/nacos/v1/cs/ops/derby";
        Query query = Query.newInstance().addParam("sql", "select * from users");
        RestResult<List<Map<String, Object>>> result = NACOS_REST_TEMPLATE.get(url, Header.EMPTY, query,
                new GenericType<RestResult<List<Map<String, Object>>>>() {
                }.getType());
        System.out.println(result.getData());
        assertTrue(result.ok());
        List<Map<String, Object>> list = result.getData();
        assertEquals(1, list.size());
        assertEquals("nacos", list.get(0).get("USERNAME"));
    }
    
    @Test
    void testDerbyOperationsNoSelect() throws Exception {
        String url = "http://127.0.0.1:8848/nacos/v1/cs/ops/derby";
        Query query = Query.newInstance().addParam("sql", "update users set username='nacos'");
        RestResult<Object> result = NACOS_REST_TEMPLATE.get(url, Header.EMPTY, query,
                new GenericType<RestResult<Object>>() {
                }.getType());
        System.out.println(result);
        assertFalse(result.ok());
        assertEquals("Only query statements are allowed to be executed", result.getMessage());
    }
    
    @Test
    void testDerbyWithErrorHandling() throws Exception {
        
        ThreadUtils.sleep(5000);
        
        boolean result = iconfig7.publishConfig("raft_test_raft_error", "cluster_test_1",
                "this.is.raft_cluster=lessspring_7");
        assertTrue(result);
        
        NotifyCenter.registerToPublisher(RaftDbErrorRecoverEvent.class, 8);
        
        CountDownLatch latch1 = new CountDownLatch(1);
        NotifyCenter.registerSubscriber(new Subscriber<RaftDbErrorEvent>() {
            @Override
            public void onEvent(RaftDbErrorEvent event) {
                latch1.countDown();
            }
            
            @Override
            public Class<? extends Event> subscribeType() {
                return RaftDbErrorEvent.class;
            }
        });
        NotifyCenter.publishEvent(new RaftDbErrorEvent());
        latch1.await(10_000L, TimeUnit.MILLISECONDS);
        
        result = iconfig7.publishConfig("raft_test_raft_error", "cluster_test_1", "this.is.raft_cluster=lessspring_7");
        assertFalse(result);
        
        CountDownLatch latch2 = new CountDownLatch(1);
        NotifyCenter.registerSubscriber(new Subscriber<RaftDbErrorRecoverEvent>() {
            
            @Override
            public void onEvent(RaftDbErrorRecoverEvent event) {
                latch2.countDown();
            }
            
            @Override
            public Class<? extends Event> subscribeType() {
                return RaftDbErrorRecoverEvent.class;
            }
        });
        NotifyCenter.publishEvent(new RaftDbErrorRecoverEvent());
        latch2.await(10_000L, TimeUnit.MILLISECONDS);
        
        result = iconfig7.publishConfig("raft_test_raft_error", "cluster_test_1", "this.is.raft_cluster=lessspring_7");
        assertTrue(result);
    }
    
    @Test
    void testIdGeneratorLeaderTransfer() throws Exception {
        ConfigurableApplicationContext context7 = applications.get("8847");
        ConfigurableApplicationContext context8 = applications.get("8848");
        ConfigurableApplicationContext context9 = applications.get("8849");
        IdGeneratorManager manager7 = context7.getBean(IdGeneratorManager.class);
        IdGeneratorManager manager8 = context8.getBean(IdGeneratorManager.class);
        IdGeneratorManager manager9 = context9.getBean(IdGeneratorManager.class);
        
        CPProtocol protocol7 = context7.getBean(CPProtocol.class);
        CPProtocol protocol8 = context8.getBean(CPProtocol.class);
        CPProtocol protocol9 = context9.getBean(CPProtocol.class);
        
        final String configGroup = PersistenceConstant.CONFIG_MODEL_RAFT_GROUP;
        long preId = -1L;
        
        if (protocol7.isLeader(configGroup)) {
            preId = manager7.nextId(CONFIG_INFO_ID);
        }
        if (protocol8.isLeader(configGroup)) {
            preId = manager8.nextId(CONFIG_INFO_ID);
        }
        if (protocol9.isLeader(configGroup)) {
            preId = manager9.nextId(CONFIG_INFO_ID);
        }
        
        // transfer leader to ip:8807
        
        Map<String, String> transfer = new HashMap<>();
        transfer.put(JRaftConstants.TRANSFER_LEADER, InetUtils.getSelfIP() + ":9847");
        RestResult<String> result = protocol7.execute(transfer);
        System.out.println(result);
        assertTrue(result.ok());
        
        TimeUnit.SECONDS.sleep(2);
        
        assertTrue(protocol7.isLeader(configGroup));
        long currentId = manager7.nextId(CONFIG_INFO_ID);
        assertNotEquals(preId, currentId);
        preId = currentId;
        
        // transfer leader to ip:8808
        
        transfer = new HashMap<>();
        transfer.put(JRaftConstants.TRANSFER_LEADER, InetUtils.getSelfIP() + ":9848");
        result = protocol8.execute(transfer);
        System.out.println(result);
        assertTrue(result.ok());
        
        TimeUnit.SECONDS.sleep(2);
        
        assertTrue(protocol8.isLeader(configGroup));
        currentId = manager8.nextId(CONFIG_INFO_ID);
        assertNotEquals(preId, currentId);
        preId = currentId;
        
        // transfer leader to ip:8809
        
        transfer = new HashMap<>();
        transfer.put(JRaftConstants.TRANSFER_LEADER, InetUtils.getSelfIP() + ":9849");
        result = protocol9.execute(transfer);
        System.out.println(result);
        assertTrue(result.ok());
        
        TimeUnit.SECONDS.sleep(2);
        
        assertTrue(protocol9.isLeader(configGroup));
        currentId = manager9.nextId(CONFIG_INFO_ID);
        assertNotEquals(preId, currentId);
    }
    
}
