package com.deloitte.system.job;

import cn.hutool.core.collection.CollUtil;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.common.core.constant.GlobalConst;
import com.common.core.enums.TableNameEnum;
import com.common.core.enums.YesNoEnum;
import com.common.core.exception.BizException;
import com.common.core.service.SFService;
import com.common.core.utils.DistributedLock;
import com.common.core.utils.RestUtil;
import com.common.redis.util.RedisUtil;
import com.deloitte.system.request.SFTokenDto;
import com.deloitte.system.request.TxConfirmDto;
import com.deloitte.system.service.ConfirmTxService;
import com.deloitte.system.service.FileService;
import com.jfinal.plugin.activerecord.Db;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.HttpHeaders;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.stream.Collectors;

/**
 * @author 廖振钦
 * @company deloitte
 * @date 2022-01-26
 * @descrition:
 *     兜底机制定时任务TODO
 * */
@Slf4j
@Component
public class GuaranteedTask {
    @Autowired
    private FileService fileService;
    @Autowired
    private SFService sfService;

    @Autowired
    private ConfirmTxService confirmTxService;

    @Autowired
    private RedisUtil redisUtil;

    @Autowired
    private DistributedLock distributedLock;
    @Value("${salesforce.baseUrl}")
    private String baseUrl;
    /**
     * 每分钟跑一次
     */
    @Scheduled(cron = "0 * * * * ?")
    public void execute() {
        boolean isLock = false;
        boolean masterisLock = distributedLock.getLock("guaranteedLock-master",3600 * 1000);
        if(!masterisLock){
            isLock = distributedLock.getLock("guaranteedLock-slave",3600 * 1000);
        }
        if(masterisLock || isLock) {  //确保两台服务器都能同时跑又不会重复跑
            try {
                List<String> sets = new ArrayList<>();
                Set<String> insertsets = redisUtil.getKeys(GlobalConst.PIPL_UNCONFIRM_INSERT + "*");
                Set<String> updatesets = redisUtil.getKeys(GlobalConst.PIPL_UNCONFIRM_UPDATE + "*");
                sets.addAll(insertsets);
                sets.addAll(updatesets);
                Map<String, String> hashmap = new HashMap<>();
                for (String key : sets) {
                    String[] keys = key.split(":");
                    hashmap.put(keys[keys.length - 1], key);
                }
                List<String> sortkeysets = hashmap.keySet().stream().sorted(new Comparator<String>(){
                    @Override
                    public int compare(String o1, String o2) {
                        Long o1L = Long.parseLong(o1);
                        Long o2L = Long.parseLong(o2);
                        Long v = o2L-o1L; //倒序
//                        Long v = o1L-o2L; //顺序
                        if(v > 0){
                            return 1;
                        }else if(v < 0){
                            return -1;
                        }else {
                            return 0;
                        }
                    }
                }).limit(50).collect(Collectors.toList());
                if (CollUtil.isEmpty(sortkeysets)) {
                    return;
                }
                // 调用SF查询事务状态的接口，传入txId List，将查询的状态结果放在map，key为txId，value为处理状态
                Map<String, Map<String,String>> statusMap = queryTransactionStatus(sortkeysets);
//                Map<String, Map<String, String>> statusMap = query();
                for (String key : sortkeysets) {
                    boolean isExecute = distributedLock.getLock("guaranteed:"+key+":lock",3600 * 1000);//判断是否有机器在执行事务
                    //如果redis里面不存在这个key与没有机器在执行该事务确认,就执行确认事务,防止多台服务器重复跑
                    if((redisUtil.getKeys(GlobalConst.PIPL_UNCONFIRM_INSERT + "*").contains(GlobalConst.PIPL_UNCONFIRM_INSERT + key)
                    || redisUtil.getKeys(GlobalConst.PIPL_UNCONFIRM_UPDATE + "*").contains(GlobalConst.PIPL_UNCONFIRM_UPDATE + key))
                    && isExecute){
                        String txid = hashmap.get(key);
                        try {
                            // 获取该TXID的处理状态，如果成功，则确认事务，如果失败则删除key
                            List<String> collect = statusMap.keySet().stream().filter(i -> i.equals(key)).collect(Collectors.toList());
                            if (CollUtil.isNotEmpty(collect)) {
                                String tId = collect.get(0);
                                Map<String, String> stringStringMap = statusMap.get(tId);
                                String status = stringStringMap.get("status");
                                String sfRecordId = stringStringMap.get("sfRecordId");
                                if ("success".equals(status)) {
                                    TxConfirmDto dto=new TxConfirmDto(key,sfRecordId, YesNoEnum.YES.getCode(),null);
                                    confirmTxService.confirm(dto);
                                }else {
                                    redisUtil.deleteKey(txid);
                                }
                            }
                        } catch (BizException e) {
                            log.error(e.getMessage(), e);
                        }finally {
                            distributedLock.releaseLock("guaranteed:"+key+":lock");
                        }
                    }
                }
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }finally {
                if(masterisLock){
                    distributedLock.releaseLock("guaranteedLock-master");
                }
                if (isLock){
                    distributedLock.releaseLock("guaranteedLock-slave");
                }
            }
        }
    }
    /**
     * 每分钟跑一次
     */
    @Scheduled(cron = "0 * * * * ?")
    public void executeFile() {
        boolean isLock = false;
        boolean masterisLock = distributedLock.getLock("guaranteedLock-master-file",3600 * 1000);
        if(!masterisLock){
            isLock = distributedLock.getLock("guaranteedLock-slave-file",3600 * 1000);
        }
        if(masterisLock || isLock) {  //确保两台服务器都能同时跑又不会重复跑
            try{
                List<String> sets = new ArrayList<>();
                Set<String> fileKeys = redisUtil.getKeys(GlobalConst.PIPL_UNCONFIRM_FILE + "*");
                sets.addAll(fileKeys);
                Map<String, String> hashmap = new HashMap<>();
                for (String key : sets) {
                    String[] keys = key.split(":");
                    hashmap.put(keys[keys.length - 1], key);
                }
                List<String> sortkeysets = hashmap.keySet().stream().sorted(new Comparator<String>(){
                    @Override
                    public int compare(String o1, String o2) {
                        Long o1L = Long.parseLong(o1);
                        Long o2L = Long.parseLong(o2);
                        Long v = o2L-o1L; //倒序
//                        Long v = o1L-o2L; //顺序
                        if(v > 0){
                            return 1;
                        }else if(v < 0){
                            return -1;
                        }else {
                            return 0;
                        }
                    }
                }).limit(50).collect(Collectors.toList());
                if (CollUtil.isEmpty(sortkeysets)) {
                    return;
                }
                Map<String, Map<String, String>> statusMap = queryTransactionStatus(sortkeysets);
//                Map<String, String> statusMap = query();
                for (String key : sortkeysets) {
                    boolean isExecute = distributedLock.getLock("guaranteed:"+key+":lock",3600 * 1000);//判断是否有机器在执行事务
                    //如果redis里面不存在这个key与没有机器在执行该事务确认,就执行确认事务,防止多台服务器重复跑
                    if((redisUtil.getKeys(GlobalConst.PIPL_UNCONFIRM_FILE + "*").contains(GlobalConst.PIPL_UNCONFIRM_FILE + key))
                            && isExecute) {
                        String txid = hashmap.get(key);
                        try {
                            //获取该TXID的处理状态，如果成功，则确认事务，如果失败则删除key
                            List<String> collect = statusMap.keySet().stream().filter(i -> i.equals(key)).collect(Collectors.toList());
                            if (CollUtil.isNotEmpty(collect)) {
                                String tId = collect.get(0);
                                Map<String, String> stringStringMap = statusMap.get(tId);
                                String status = stringStringMap.get("status");
                                String sfRecordId = stringStringMap.get("sfRecordId");
                                if ("success".equals(status)) {
//                                    confirmTxService.confirmFile(key);
                                    redisUtil.deleteKey(txid);
                                }else {
                                    List<String> keyList = redisUtil.getObjects(txid, String.class);
                                    try {
                                        fileService.cleanS3File(keyList);
                                    }catch (Exception e){
                                        log.error(e.getMessage(), e);
                                        throw new BizException("clean s3 file fail");
                                    }
                                    redisUtil.deleteKey(txid);
                                }
                            }
                        } catch (BizException e) {
                            log.error(e.getMessage(), e);
                        }finally {
                            distributedLock.releaseLock("guaranteed:"+key+":lock");
                        }
                    }
                }
            }catch (Exception e){
                log.error(e.getMessage(), e);
            }finally {
                if(masterisLock){
                    distributedLock.releaseLock("guaranteedLock-master-file");
                }
                if (isLock){
                    distributedLock.releaseLock("guaranteedLock-slave-file");
                }
            }
        }
    }


    /**
     * 同步表中的sfid(每天两点执行)
     */
    @Scheduled(cron = "0 0 2 * * ?")
    public void executeSyncSfid() {
        boolean isLock = false;
        boolean masterisLock = distributedLock.getLock("guaranteedLock-syncSfid-master",3600 * 1000);
        if(!masterisLock){
            isLock = distributedLock.getLock("guaranteedLock-syncSfid-slave",3600 * 1000);
        }
        if(masterisLock || isLock) {  //确保两台服务器都能同时跑又不会重复跑
            ArrayList<String> lockKeys = new ArrayList<>();
            try {
                //aws tableName与sf tableName的映射
                List<String> allAwsTableName = TableNameEnum.getAllAwsTableName();
                for (String awsTableName:allAwsTableName) {
                    List<Long> ids = Db.query("select id from `"+awsTableName+"` where is_delete = '0' and create_time>date_add(now(),interval -2 MONTH) and sf_record_id is null");
                    log.info("同步表{}，查询到sf_id为空的数据共{}条",awsTableName,ids.size());
                    if (CollUtil.isEmpty(ids)) {
                        continue;
                    }
                    //加表名锁，阻止多个服务同时更新一张表
                    //判断是否有机器在执行事务
                    boolean isExecute = distributedLock.getLock("guaranteed-syncSfid:"+awsTableName+":lock",3600 * 1000);
                    if(isExecute) {
                        lockKeys.add(awsTableName);
                        String sfTableName = TableNameEnum.getSfTableNameByAwsTableName(awsTableName);
                        //截取ids每50条执行一次
                        int skip = 0;
                        int sub = 50;
                        List<String> collect = null;
                        String idColumn = "AWS_Data_Id__c";
                        while (collect==null||collect.size()>=sub) {
                            collect = ids.stream().skip(skip).limit(sub).map(String::valueOf).collect(Collectors.toList());
                            skip += sub;
                            String sql = "select Id,"+idColumn+"  from "+sfTableName+" where "+idColumn+" in "+collect.stream().collect(Collectors.joining("','", "('", "')"));
                            JSONArray jsonArray = sfService.querySFData(sql);
                            if (jsonArray != null && jsonArray.size() != 0) {
                                for (int i = 0;i<jsonArray.size();i++) {
                                    JSONObject data = jsonArray.getJSONObject(i);
                                    String sfId = data.getString("Id");
                                    String awsId = data.getString(idColumn);
                                    Db.update("update `"+awsTableName+"` set sf_record_id = '"+ sfId +"' where id = "+awsId);
                                    log.info("更新{} sfid,id:{},sfid:{}",awsTableName,awsId,sfId);
                                }
                            }
                        }
                    }
                }
            } catch (Exception e) {
                log.error("同步contact sfid异常："+e.getMessage(), e);
            }finally {
                if(masterisLock){
                    distributedLock.releaseLock("guaranteedLock-syncSfid-master");
                }
                if (isLock){
                    distributedLock.releaseLock("guaranteedLock-syncSfid-slave");
                }
                //释放数据锁
                if(CollUtil.isNotEmpty(lockKeys)){
                    for (String awsTableName:lockKeys) {
                        distributedLock.releaseLock("guaranteed-syncSfid:"+awsTableName+":lock");
                    }
                }
            }
        }
    }


    public Map<String,Map<String,String>> queryTransactionStatus(List<String> sortkeysets){
        SFTokenDto tokenDto = sfService.getToken();
        String accessToken = tokenDto.getAccessToken();
        HttpHeaders header = new HttpHeaders();
        header.add("Authorization", "Bearer " + accessToken);
        StringBuilder idString = new StringBuilder();
        // 拼接URL
        String ids = sortkeysets.stream().collect(Collectors.joining("','", "('", "')"));
        idString.append(baseUrl).append("services/data/v53.0/query/?q=Select Status__c,TransId__c,SFRecordId__c from Transaction_Log__c Where TransId__c in ").append(ids);
        // 请求数据
        JSONObject jsonObject = RestUtil.get(idString.toString(),header);
        Map<String, Map<String,String>> map = new HashMap<>();
        List<JSONObject> objects = new ArrayList<>();
        Object totalSize = jsonObject.get("totalSize");
        Boolean done =(Boolean) jsonObject.get("done");
        String nextRecordsUrl = jsonObject.getString("nextRecordsUrl");
        JSONArray jsonArray = jsonObject.getJSONArray("records");

        if (totalSize == null) {
            log.info("请求事务确认数据失败,result:{}",jsonObject);
            throw new BizException("请求事务确认数据失败");
        }
        for (int i = 0; i <jsonArray.size() ; i++) {
            JSONObject data = jsonArray.getJSONObject(i);
            String status = data.getString("Status__c");
            String transId = data.getString("TransId__c");
            String sfRecordId = data.getString("SFRecordId__c");
            Map<String, String> mapStatus = new HashMap<>();
            mapStatus.put("status",status);
            mapStatus.put("sfRecordId",sfRecordId);
            map.put(transId,mapStatus);
        }
        return map;
    }

    public Map<String, String> queryNextRecords(String nextRecordsUrl,Map<String,String> map){
        SFTokenDto tokenDto = sfService.getToken();
        String accessToken = tokenDto.getAccessToken();
        HttpHeaders header = new HttpHeaders();
        header.add("Authorization", "Bearer " + accessToken);
        JSONObject jsonObject = RestUtil.get(nextRecordsUrl,header);
        Boolean done =(Boolean) jsonObject.get("done");
        JSONArray jsonArray = jsonObject.getJSONArray("records");
        Object totalSize = jsonObject.get("totalSize");
        if (totalSize == null) {
            return map;
        }
        for (int i = 0; i <jsonArray.size() ; i++) {
            JSONObject data = jsonArray.getJSONObject(i);
            String status = data.getString("Status__c");
            String transId = data.getString("TransId__c");
            map.put(transId,status);
        }
        return map;
    }
}