/*
 * 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.config.server.service.dump.processor;

import com.alibaba.nacos.common.task.NacosTask;
import com.alibaba.nacos.common.task.NacosTaskProcessor;
import com.alibaba.nacos.common.utils.MD5Utils;
import com.alibaba.nacos.common.utils.StringUtils;
import com.alibaba.nacos.config.server.model.ConfigInfoWrapper;
import com.alibaba.nacos.config.server.service.ClientIpWhiteList;
import com.alibaba.nacos.config.server.service.ConfigCacheService;
import com.alibaba.nacos.config.server.service.SwitchService;
import com.alibaba.nacos.config.server.service.dump.task.DumpAllTask;
import com.alibaba.nacos.config.server.service.repository.ConfigInfoPersistService;
import com.alibaba.nacos.config.server.utils.GroupKey2;
import com.alibaba.nacos.config.server.utils.LogUtil;
import com.alibaba.nacos.config.server.utils.PropertyUtil;
import com.alibaba.nacos.api.model.Page;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import static com.alibaba.nacos.config.server.constant.Constants.ENCODE_UTF8;
import static com.alibaba.nacos.config.server.utils.LogUtil.DEFAULT_LOG;

/**
 * Dump all processor.
 *
 * @author Nacos
 * @date 2020/7/5 12:19 PM
 */
public class DumpAllProcessor implements NacosTaskProcessor {
    
    public DumpAllProcessor(ConfigInfoPersistService configInfoPersistService) {
        this.configInfoPersistService = configInfoPersistService;
    }
    
    @Override
    @SuppressWarnings("PMD.MethodTooLongRule")
    public boolean process(NacosTask task) {
        if (!(task instanceof DumpAllTask)) {
            DEFAULT_LOG.error(
                    "[all-dump-error] ,invalid task type {},DumpAllProcessor should process DumpAllTask type.",
                    task.getClass().getSimpleName());
            return false;
        }
        DumpAllTask dumpAllTask = (DumpAllTask) task;
        
        long currentMaxId = configInfoPersistService.findConfigMaxId();
        long lastMaxId = 0;
        ThreadPoolExecutor executorService = null;
        if (dumpAllTask.isStartUp()) {
            executorService = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(),
                    Runtime.getRuntime().availableProcessors(), 60L, TimeUnit.SECONDS,
                    new LinkedBlockingQueue<>(PropertyUtil.getAllDumpPageSize() * 2),
                    r -> new Thread(r, "dump all executor"), new ThreadPoolExecutor.CallerRunsPolicy());
        } else {
            executorService = new ThreadPoolExecutor(1, 1, 60L, TimeUnit.SECONDS, new SynchronousQueue<>(),
                    r -> new Thread(r, "dump all executor"), new ThreadPoolExecutor.CallerRunsPolicy());
        }
        
        DEFAULT_LOG.info("start dump all config-info...");
        
        while (lastMaxId < currentMaxId) {
            long start = System.currentTimeMillis();
            
            Page<ConfigInfoWrapper> page = configInfoPersistService.findAllConfigInfoFragment(lastMaxId,
                    PropertyUtil.getAllDumpPageSize(), dumpAllTask.isStartUp());
            long dbTimeStamp = System.currentTimeMillis();
            if (page == null || page.getPageItems() == null || page.getPageItems().isEmpty()) {
                break;
            }
            
            for (ConfigInfoWrapper cf : page.getPageItems()) {
                lastMaxId = Math.max(cf.getId(), lastMaxId);
                if (StringUtils.isBlank(cf.getTenant())) {
                    continue;
                }
                //if not start up, page query will not return content, check md5 and lastModified first ,if changed ,get single content info to dump.
                if (!dumpAllTask.isStartUp()) {
                    final String groupKey = GroupKey2.getKey(cf.getDataId(), cf.getGroup(), cf.getTenant());
                    boolean newLastModified = cf.getLastModified() > ConfigCacheService.getLastModifiedTs(groupKey);
                    //check md5 & update local disk cache.
                    String localContentMd5 = ConfigCacheService.getContentMd5(groupKey);
                    boolean md5Update = !localContentMd5.equals(cf.getMd5());
                    if (newLastModified || md5Update) {
                        LogUtil.DUMP_LOG.info("[dump-all] find change config {}, {}, md5={}", groupKey,
                                cf.getLastModified(), cf.getMd5());
                        cf = configInfoPersistService.findConfigInfo(cf.getDataId(), cf.getGroup(), cf.getTenant());
                    } else {
                        continue;
                    }
                }
                
                if (cf == null) {
                    continue;
                }
                
                if (cf.getDataId().equals(ClientIpWhiteList.CLIENT_IP_WHITELIST_METADATA)) {
                    ClientIpWhiteList.load(cf.getContent());
                }
                
                if (cf.getDataId().equals(SwitchService.SWITCH_META_DATA_ID)) {
                    SwitchService.load(cf.getContent());
                }
                
                final String content = cf.getContent();
                final String dataId = cf.getDataId();
                final String group = cf.getGroup();
                final String tenant = cf.getTenant();
                final long lastModified = cf.getLastModified();
                final String type = cf.getType();
                final String encryptedDataKey = cf.getEncryptedDataKey();
                
                executorService.execute(() -> {
                    final String md5Utf8 = MD5Utils.md5Hex(content, ENCODE_UTF8);
                    boolean result = ConfigCacheService.dumpWithMd5(dataId, group, tenant, content, md5Utf8,
                            lastModified, type, encryptedDataKey);
                    if (result) {
                        LogUtil.DUMP_LOG.info("[dump-all-ok] {}, {}, length={},md5UTF8={}",
                                GroupKey2.getKey(dataId, group), lastModified, content.length(), md5Utf8);
                    } else {
                        LogUtil.DUMP_LOG.info("[dump-all-error] {}", GroupKey2.getKey(dataId, group));
                    }
                    
                });
                
            }
            
            long diskStamp = System.currentTimeMillis();
            DEFAULT_LOG.info("[all-dump] submit all task for {} / {}, dbTime={},diskTime={}", lastMaxId, currentMaxId,
                    (dbTimeStamp - start), (diskStamp - dbTimeStamp));
        }
        
        //wait all task are finished and then shutdown executor.
        try {
            int unfinishedTaskCount = 0;
            while ((unfinishedTaskCount = executorService.getQueue().size() + executorService.getActiveCount()) > 0) {
                DEFAULT_LOG.info("[all-dump] wait {} dump tasks to be finished", unfinishedTaskCount);
                Thread.sleep(1000L);
            }
            executorService.shutdown();
            
        } catch (Exception e) {
            DEFAULT_LOG.error("[all-dump] wait  dump tasks to be finished error", e);
        }
        DEFAULT_LOG.info("success to  dump all config-info。");
        return true;
    }
    
    final ConfigInfoPersistService configInfoPersistService;
}
