package com.product.data.center.service;

import cn.hutool.core.codec.Base64;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.collection.ListUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.io.file.FileWriter;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.*;
import cn.hutool.crypto.SecureUtil;
import cn.hutool.crypto.symmetric.AES;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.product.core.cache.util.RedisUtil;
import com.product.core.config.CoreConst;
import com.product.core.dao.BaseDao;
import com.product.core.entity.DataTableEntity;
import com.product.core.entity.FieldSetEntity;
import com.product.core.exception.BaseException;
import com.product.core.spring.context.SpringMVCContextHolder;
import com.product.data.center.config.CmnConst;
import com.product.data.center.config.ErrorCode;
import com.product.data.center.utils.QuerySqlParseUtil;
import com.product.datasource.config.DataBaseType;
import com.product.datasource.dao.Dao;
import com.product.datasource.entity.DataBaseEntity;
import com.product.util.BaseUtil;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;
import redis.clients.jedis.Jedis;

import javax.annotation.Resource;
import java.io.File;
import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedTransferQueue;
import java.util.stream.Collectors;

/**
 * @Author cheng
 * @Date 2022/11/29 8:29
 * @Desc 同步删除记录
 */
@Service
public class SyncDelRecordService {
    @Resource
    private BaseDao baseDao;


    public void runTask() {
        DataTableEntity configDt = baseDao.listTable(CmnConst.PRODUCT_MES_DEL_RECORD_CONFIG);
        if (DataTableEntity.isEmpty(configDt)) {
            return;
        }
        for (int i = 0; i < configDt.getRows(); i++) {
            TriggerDeleteService triggerDeleteService =null;
            try {
                triggerDeleteService = new TriggerDeleteService(configDt.getFieldSetEntity(i));
                    if ("2".equals(triggerDeleteService.getDelType())) {
                        //子库采集表删除
                        triggerDeleteService.collectTableDelete();
                    } else {
                        //主库表删除
                        triggerDeleteService.mainDataBaseTableDelete();
                    }
            } catch (Exception e) {
                if(triggerDeleteService!=null){
                    Set<Dao> daoSet = triggerDeleteService.getDaoSet();
                    for (Dao dao : daoSet) {
                        dao.closeConnection();
                    }
                }
                e.printStackTrace();
                SpringMVCContextHolder.getSystemLogger().error(e);
            }
        }
        System.out.println("完成");
    }

    private static ExecutorService singleThreadExecutor = null;

    private static Queue<JSONObject> queue = null;

    public static synchronized void WriteData(List<Record> records, String fileName) {
        if (CollectionUtil.isEmpty(records)) {
            return;
        }
        if (queue == null) {
            queue = new LinkedTransferQueue<>();
        }
        if (StringUtils.isEmpty(fileName)) {
            throw new BaseException(ErrorCode.FILENAME_CAN_NOT_EMPTY);
        }
        List<Object> collect = ListUtil.toList(records.stream().map(item -> new JSONObject((Map) item.getRecord().getValues())).toArray());
        JSONArray objects = new JSONArray(collect);
        JSONObject jsonObject = new JSONObject();
        jsonObject.put(com.product.file.config.CmnConst.FILE_NAME, fileName);
        jsonObject.put("data", objects);

        byte[] key = Base64.decode("glwqQbMzmInerCSs9RoAWQ==");
        AES aes = SecureUtil.aes(key);
        queue.add(jsonObject);
        if (singleThreadExecutor != null && !singleThreadExecutor.isTerminated()) {
            return;
        }
        singleThreadExecutor = Executors.newSingleThreadExecutor();
//        if(singleThreadExecutor.)
        singleThreadExecutor.submit(() -> {
            while (true) {
                if (queue.isEmpty()) {
                    ThreadUtil.sleep(30000);
                    continue;
                }
                try {
                    JSONObject poll = queue.poll();
                    String name = poll.getString(com.product.file.config.CmnConst.FILE_NAME);
                    JSONArray data = poll.getJSONArray("data");
                    if (StringUtils.isEmpty(name) || (data == null || data.isEmpty())) {
                        SpringMVCContextHolder.getSystemLogger().error(new BaseException(ErrorCode.FILENAME_OR_DATA_EMPTY));
                        continue;
                    }
                    String now = DateUtil.format(new Date(), "yyyyMMdd");
                    String filePath = "." + File.separator + "触发器删除数据记录" + File.separator + now + File.separator + name + ".dbackups";
                    File file = new File(filePath);
                    File parentFile = file.getParentFile();
                    if (!parentFile.exists()) {
                        parentFile.mkdirs();
                    }
                    if (!file.exists()) {
                        file.createNewFile();
                    }
                    FileWriter writer = new FileWriter(file);
                    writer.append(aes.encryptBase64(data.toJSONString()));
                    writer.append("\n");
                    System.out.println("写入成功");
                } catch (Exception e) {
                    e.printStackTrace();
                    SpringMVCContextHolder.getSystemLogger().error(e);
                }
            }
        });
        singleThreadExecutor.shutdown();
    }

    public static class Record {

        private int deleteFlag = 1;
        private Integer delTableId;

        private String[] uniqueValue;

        private String sourceInfo;

        private String[] primaryValue;

        private FieldSetEntity record;

        public Record(Integer delTableId, String[] uniqueValue) {
            this.delTableId = delTableId;
            this.uniqueValue = uniqueValue;
        }

        public Record(Integer delTableId, String[] uniqueValue, String sourceInfo) {
            this.delTableId = delTableId;
            this.uniqueValue = uniqueValue;
            this.sourceInfo = sourceInfo;
        }

        public Record(String uniqueValue[], String sourceInfo) {
            this.uniqueValue = uniqueValue;
            this.sourceInfo = sourceInfo;
        }

        public void setRecord(FieldSetEntity record) {
            this.record = record;
        }

        public void setDeleteFlag(int deleteFlag) {
            this.deleteFlag = deleteFlag;
        }

        public FieldSetEntity getRecord() {
            return record;
        }

        public int getDeleteFlag() {
            return deleteFlag;
        }

        public Integer getDelTableId() {
            return delTableId;
        }

        public String[] getUniqueValue() {
            return uniqueValue;
        }

        public void setUniqueValue(String[] uniqueValue) {
            this.uniqueValue = uniqueValue;
        }

        public String getSourceInfo() {
            return sourceInfo;
        }

        public String[] getPrimaryValue() {
            return primaryValue;
        }

        public void setPrimaryValue(String[] primaryValue) {
            this.primaryValue = primaryValue;
        }


        @Override
        public boolean equals(Object o) {
            if (this == o) return true;
            if (!(o instanceof Record)) return false;
            Record record = (Record) o;
            return ArrayUtil.equals(uniqueValue, record.uniqueValue) && (Objects.equals(sourceInfo, record.sourceInfo));
        }

        @Override
        public int hashCode() {
            return Objects.hash(uniqueValue, sourceInfo);
        }
    }


    private class TriggerDeleteService {
        //删除记录数据源
        private String dataSource;
        //删除目标数据源
        private String[] targetDataSource;
        //表名
        private String tableName;
        //随机后缀
        private String randomSuffix;
        //唯一字段名称（多个组合唯一）
        private String[] uniqueFields;
        //子库存储主库唯一字段名称
        private String preMasterField;
        //删除记录表名称
        private String delRecordTable;
        //采集表id
        private String collectId;
        //采集来源记录字段
        private String sourceInfoField;
        // 删除类型 1 主库删除后同步删除子库 2 子库删除后同步到主库
        private String delType;
        //删除表数据记录
        private DataTableEntity deleteTableRecord;

        private Dao sourceDao;

        private String targetTable;

        public String getDelType() {
            return delType;
        }

        private Set<Dao> daoSet=new HashSet<>();

        public Set<Dao> getDaoSet() {
            return daoSet;
        }

        public TriggerDeleteService(FieldSetEntity fse) throws SQLException {
            //数据源
            this.dataSource = fse.getString("data_source");
            //目标数据源
            this.targetDataSource = fse.getString("target_data_source").split(",");
            //表名
            this.tableName = StringUtils.lowerCase(fse.getString("table_name"));
            //随机后缀
            this.randomSuffix = fse.getString("random_suffix");
            //唯一字段
            this.uniqueFields = StringUtils.lowerCase(fse.getString("unique_field")).split(",");
            //主库存储子库字段
            this.preMasterField = StringUtils.lowerCase(fse.getString("pre_master_field"));
            //删除记录表表名
            this.delRecordTable = "trigger_table_" + this.randomSuffix;
            // 删除类型 1 主库删除后同步删除子库 2 子库删除后同步到主库
            this.delType = fse.getString("del_type");
            //采集id
            this.collectId = fse.getString("collect_id");
            //采集ID字段
            this.sourceInfoField = StringUtils.lowerCase(fse.getString("collect_field"));
            if ("2".equals(this.getDelType())) {
                //
                StringBuilder sql = new StringBuilder();
                sql.append("\n SELECT extract.* ");
                sql.append("\n FROM ");
                sql.append("\n product_sys_data_collect collect ");
                sql.append("\n JOIN product_sys_data_extract_config extract ON ");
                sql.append("\n  upper( case when length(collect.target_table)>0 then ");
                sql.append("\n collect.target_table else collect.source_table end ) = upper( extract.extract_source_table )");
                sql.append(" where collect.uuid=? limit 1");

                FieldSetEntity fs = baseDao.getFieldSetEntityBySQL(sql.toString(), new Object[]{fse.getString(CmnConst.CONFIG_UUID)}, false);
                if (fs == null) {
                    throw new RuntimeException("提取配置不存在");
                }
                this.targetTable = fs.getString("extract_target_table");
            }
            if (StringUtils.isEmpty(targetTable)) {
                this.targetTable = tableName;
            }
            this.targetTable = StringUtils.lowerCase(targetTable);
            DataBaseEntity dbe = new DataBaseEntity(this.dataSource);
            sourceDao = dbe.getDao();
            daoSet.add(sourceDao);
            Connection connection = sourceDao.getConnection();
            connection.setAutoCommit(false);
            try {
                selectDelRecord(fse.getUUID());

            } catch (Exception e) {
                connection.close();
                throw e;
            }
        }

        /**
         * 采集表删除同步到主
         */
        public void collectTableDelete() throws SQLException {
            Dao targetDao = null;
            try {
                if (DataTableEntity.isEmpty(this.deleteTableRecord)) {
                    return;
                }
                List<FieldSetEntity> data = this.deleteTableRecord.getData();
                List<List<FieldSetEntity>> splitRecord = CollectionUtil.split(data, 1000);
                for (List<FieldSetEntity> fieldSetEntityList : splitRecord) {
                    Map<String, List<Record>> groupUniqueValues = new HashMap<>();
                    fieldSetEntityList.stream().forEach(item -> {
                        List<Record> records = groupUniqueValues.computeIfAbsent(item.getString(this.sourceInfoField), k -> new ArrayList<>());
                        //子库的主键字段值
                        String[] uniqueValues = new String[this.uniqueFields.length];
                        for (int i = 0; i < this.uniqueFields.length; i++) {
                            uniqueValues[i] = item.getString(this.uniqueFields[i]);
                        }
                        Record record = new Record(item.getInteger(("ID_" + this.randomSuffix).toLowerCase()), uniqueValues, item.getString(this.sourceInfoField));
                        record.setDeleteFlag("2".equals(item.getString("delete_flag")) ? 2 : 1);
                        records.add(record);
                    });

                    if (groupUniqueValues.isEmpty()) {
                        return;
                    }
                    StringBuilder sql = new StringBuilder(128);
                    int archivingCount = 0;
                    for (List<Record> value : groupUniqueValues.values()) {
                        List<Record> collect = value.stream().filter(item -> item.getDeleteFlag() != 2).collect(Collectors.toList());
                        archivingCount += value.size() - collect.size();
                        if (!CollectionUtil.isEmpty(collect)) {
                            getUniqueValues(collect, sql);
                        }
                    }
                    if (sql.length() <= 0 && archivingCount == 0) {
                        return;
                    }
                    DataBaseEntity dbe = new DataBaseEntity(targetDataSource[0]);
                    targetDao = dbe.getDao();
                    daoSet.add(targetDao);
                    targetDao.getConnection().setAutoCommit(false);
                    if (sql.length() > 0) {

                        DataTableEntity dt = targetDao.getList(sql.toString(), (Object[]) null);
                        if (DataTableEntity.isEmpty(dt) && archivingCount == 0) {
                            return;
                        }
                        for (int i = 0; i < dt.getRows(); i++) {
                            //主库存的子库主键值
                            String preMaterKey = dt.getString(i, this.preMasterField);
                            //采集来源值
                            String sourceInfo = dt.getString(i, this.sourceInfoField);
                            List<Record> records = groupUniqueValues.get(sourceInfo);
                            if (CollectionUtil.isEmpty(records)) {
                                continue;
                            }
                            String[] primaryValue = new String[this.uniqueFields.length];
                            for (int i1 = 0; i1 < this.uniqueFields.length; i1++) {
                                primaryValue[i1] = dt.getString(i, this.uniqueFields[i1]);
                            }
                            int index = records.indexOf(new Record(new String[]{preMaterKey}, sourceInfo));
                            if (index < 0) {
                                continue;
                            }
                            Record record = records.get(index);
                            record.setPrimaryValue(primaryValue);
                            record.setRecord(dt.getFieldSetEntity(i));
                        }
                    }

                    sql.setLength(0);
                    sql.append(" DELETE FROM ").append(this.targetTable).append(" WHERE ");
                    int length = sql.length();
                    List<Record> delRecords = new ArrayList<>();
                    List<Record> lastRecord = new ArrayList<>();
                    if (archivingCount == 0 || archivingCount != this.deleteTableRecord.getRows()) {
                        for (List<Record> value : groupUniqueValues.values()) {
                            String sourceInfo = value.get(0).getSourceInfo();
                            String[] collect = value.stream().filter(item -> {
                                if (item.getDeleteFlag() == 2) {
                                    lastRecord.add(item);
                                }
                                if (ArrayUtil.isEmpty(item.getPrimaryValue())) {
                                    return false;
                                }
                                delRecords.add(item);
                                return true;
                            }).flatMap(item -> Arrays.stream(item.getPrimaryValue())).toArray(String[]::new);
                            if (ArrayUtil.isEmpty(collect)) {
                                continue;
                            }
                            if (length < sql.length()) {
                                sql.append(" OR ");
                            }
                            sql.append(" (").append(this.sourceInfoField).append("='").append(sourceInfo).append("' and (");
                            for (int i = 0; i < this.uniqueFields.length; i++) {
                                if (i > 0) {
                                    sql.append(" AND ");
                                }
                                sql.append(BaseUtil.buildQuestionMarkFilter(uniqueFields[i], collect, true));
                            }
                            sql.append(" ) )");
                        }
                        if (lastRecord.size() > 0) {
                            delRecords.removeAll(lastRecord);
                        }
                    } else {
                        lastRecord.addAll(groupUniqueValues.values().stream().flatMap(item -> item.stream()).collect(Collectors.toList()));
                    }
                    if (sql.length() > length) {
                        targetDao.executeSql(sql.toString());
                    }
                    delRecords.addAll(lastRecord);
                    List<Record> allRecord = new ArrayList<>(delRecords);

//                List<Record> records = new ArrayList<>(allRecord);
                    this.deleteArchivingData(allRecord);
                    delRecords.stream().forEach(item -> item.setDeleteFlag(1));
                    if (!CollectionUtil.isEmpty(allRecord)) {
                        //标记为归档未删除
                        allRecord.stream().forEach(item -> item.setDeleteFlag(2));
                    }
                    List<Record> collect = lastRecord.stream().filter(item -> item.getRecord() == null).collect(Collectors.toList());
                    delRecords.removeAll(collect);
                    updateRecordTableStatus(delRecords);
                    targetDao.getConnection().commit();
                    WriteData(delRecords, targetDataSource[0] + "&_" + tableName + "&_collect");
                    this.sourceDao.getConnection().commit();
                }
            } catch (Exception e) {
                throw e;
            } finally {
                if (targetDao != null) {
                    targetDao.closeConnection();
                }
                if (this.sourceDao != null) {
                    this.sourceDao.closeConnection();
                }
            }
        }

        /**
         * 主库删除同步到子
         */
        public void mainDataBaseTableDelete() throws SQLException {
            DataBaseEntity[] dataBaseEntities = null;

            try {
                if (DataTableEntity.isEmpty(this.deleteTableRecord)) {
                    return;
                }
                List<Record> recordList = new ArrayList<>();
                List<FieldSetEntity> data = this.deleteTableRecord.getData();
                List<List<FieldSetEntity>> splitRecord = CollectionUtil.split(data, 1000);
                for (List<FieldSetEntity> fieldSetEntityList : splitRecord) {
                    fieldSetEntityList.stream().forEach(item -> {
                        String[] uniqueValues = new String[this.uniqueFields.length];
                        for (int i = 0; i < this.uniqueFields.length; i++) {
                            uniqueValues[i] = item.getString(this.uniqueFields[i]);
                        }
                        if (ArrayUtil.isAllEmpty(uniqueValues)) {
                            return;
                        }
                        Record record = new Record(item.getInteger((this.randomSuffix + "_ID").toLowerCase()), uniqueValues);
                        record.setPrimaryValue(uniqueValues);
                        record.setDeleteFlag("2".equals(item.getString("delete_flag")) ? 2 : 1);
                        recordList.add(record);
                    });
                    dataBaseEntities = new DataBaseEntity[this.targetDataSource.length];
                    for (int i = 0; i < this.targetDataSource.length; i++) {
                        dataBaseEntities[i] = new DataBaseEntity(this.targetDataSource[i]);
                        Dao targetDao = dataBaseEntities[i].getDao();
                        daoSet.add(targetDao);
                        Connection connection = targetDao.getConnection();
                        connection.setAutoCommit(false);
                        DataTableEntity dt;
                        if (this.uniqueFields.length == 1) {
                            dt = targetDao.getList(tableName, BaseUtil.buildQuestionMarkFilter(this.uniqueFields[0], recordList.stream().map(item -> item.getUniqueValue()[0]).toArray(), true), null, null);
                        } else {
                            if (recordList.size() > 1000) {
                                List<List<Record>> split = ListUtil.split(recordList, 500);
                                dt = new DataTableEntity();
                                for (int j = 0; j < split.size(); j++) {
                                    dt.addFieldSetEntity(targetDao.getList(tableName, getMultiplePrimaryFieldFilter(split.get(j)), null, null));
                                }
                            } else {
                                dt = targetDao.getList(tableName, getMultiplePrimaryFieldFilter(recordList), null, null);
                            }

                        }
                        for (Record record : recordList) {
                            int deleteFlag = record.getDeleteFlag();
                            if (2 == deleteFlag) {
                                if (dt == null) {
                                    dt = new DataTableEntity();
                                }
                                FieldSetEntity fs = new FieldSetEntity();
                                fs.setTableName("temp");
                                for (int j = 0; j < this.uniqueFields.length; j++) {
                                    fs.setValue(this.uniqueFields[j], record.getUniqueValue()[j]);
                                }
                                dt.addFieldSetEntity(fs);
                            }
                        }
                        if (DataTableEntity.isEmpty(dt)) {
                            continue;
                        }
//                    List<Record> records = recordList;
                        //在数据库查询到的视为全部能删除
//                    if (dt.getRows() != recordList.size()) {

//                    }
                        if (CollectionUtil.isEmpty(recordList)) {
                            return;
                        }
                        List<Record> records = new ArrayList<>();
                        Record record;
                        for (int j = 0; j < dt.getRows(); j++) {
                            String[] uniqueValues = new String[this.uniqueFields.length];
                            for (int k = 0; k < this.uniqueFields.length; k++) {
                                uniqueValues[k] = dt.getString(j, this.uniqueFields[k]);
                            }
                            record = new Record(uniqueValues, null);
                            int index = recordList.indexOf(record);
                            if (index == -1) {
                                continue;
                            }
                            record = recordList.get(index);
                            record.setRecord(dt.getFieldSetEntity(j));
                            records.add(record);
                        }
                        if (CollectionUtil.isEmpty(records)) {
                            return;
                        }
                        List<Record> recordSub = records.stream().filter(item -> item.getDeleteFlag() != 2 && item.getRecord() != null).collect(Collectors.toList());
                        if (!CollectionUtil.isEmpty(recordSub)) {
                            if (this.uniqueFields.length == 1) {
                                targetDao.delete(tableName, BaseUtil.buildQuestionMarkFilter(this.uniqueFields[0], recordSub.stream().map(item -> item.getUniqueValue()[0]).toArray(), true));
                            } else {
                                if (records.size() > 1000) {
                                    List<List<Record>> split = ListUtil.split(recordSub, 500);
                                    for (int j = 0; j < split.size(); j++) {
                                        targetDao.delete(tableName, getMultiplePrimaryFieldFilter(split.get(j)));
                                    }
                                } else {
                                    targetDao.delete(tableName, getMultiplePrimaryFieldFilter(recordSub));
                                }
                            }
                        }
                        WriteData(recordSub, dataBaseEntities[i].getUuid() + "&_" + tableName + "&_master");
                        List<Record> deRecords = new ArrayList<>(records);
                        deleteArchivingData(deRecords);
                        records.stream().forEach(item -> item.setDeleteFlag(1));
                        if (!CollectionUtil.isEmpty(deRecords)) {
                            //标记为归档未删除
                            deRecords.stream().forEach(item -> item.setDeleteFlag(2));
                        }
                        updateRecordTableStatus(records);
                    }
                    for (DataBaseEntity dataBaseEntity : dataBaseEntities) {
                        Dao dao=dataBaseEntity.getDao();
                        daoSet.add(dao);
                        Connection connection = dao.getConnection();
                        connection.commit();
                    }
                }

                this.sourceDao.getConnection().commit();
            } catch (Exception e) {
                throw e;
            } finally {
                if (!ArrayUtil.isEmpty(dataBaseEntities)) {
                    for (DataBaseEntity dataBaseEntity : dataBaseEntities) {
                        dataBaseEntity.getDao().closeConnection();
                    }
                }
                this.sourceDao.closeConnection();
            }
        }

        /**
         * 更新触发器删除记录状态
         *
         * @param recordList
         */
        private void updateRecordTableStatus(List<Record> recordList) {
            if (CollectionUtil.isEmpty(recordList)) {
                return;
            }
            Map<Integer, List<Integer>> collect = recordList.stream().collect(Collectors.groupingBy(item -> Integer.valueOf(item.getDeleteFlag()), Collectors.mapping(item -> item.getDelTableId(), Collectors.toList())));
            for (Map.Entry<Integer, List<Integer>> entry : collect.entrySet()) {
                List<Integer> value = entry.getValue();
                value.add(0, entry.getKey());
                this.sourceDao.executeSql("update " + this.delRecordTable + " set DELETE_FLAG=? WHERE " + BaseUtil.buildQuestionMarkFilter("ID_" + this.randomSuffix, entry.getValue().size() - 1, true), entry.getValue().toArray());
            }
        }


        /**
         * 删除归档库
         *
         * @param list
         * @return
         * @throws SQLException
         */
        private void deleteArchivingData(List<Record> list) throws SQLException {
            if (CollectionUtil.isEmpty(list)) {
                return;
            }
            List<Record> recordList = new ArrayList<>(new HashSet<>(list));

            FieldSetEntity fse = baseDao.getFieldSetByFilter(CmnConst.DATA_ARCHIVING_TABLE, "UPPER(source_table)=Upper(?)", new Object[]{targetTable}, false);
            if (FieldSetEntity.isEmpty(fse)) {
                return;
            }
            String targetDataSource = fse.getString(CmnConst.TARGET_DATA_SOURCE);
            String targetTablePrefix = fse.getString("target_table_prefix");
//            String targetDataSource = "c94c19cf-86a8-4b84-85ed-3f3f90d7ecca";
//            String targetTablePrefix = "da_" + this.tableName;
            DataBaseEntity dbe = new DataBaseEntity(targetDataSource);
            Dao targetDao = dbe.getDao();
            daoSet.add(targetDao);
            Connection connection = null;
            try {
                connection = targetDao.getConnection();
                connection.setAutoCommit(false);
                Set<String> allTableName = QuerySqlParseUtil.getAllTableName(targetDao, dbe.getDbName(), targetTablePrefix);
                if (CollectionUtil.isEmpty(allTableName)) {
                    allTableName = new HashSet<>();
                    allTableName.add(targetTablePrefix);
                }
                StringBuilder lastFilter = null;
                List<Record> lastNotDeleteRecord = recordList.stream().filter(item -> item.getDeleteFlag() == 2).collect(Collectors.toList());
                if (!CollectionUtil.isEmpty(lastNotDeleteRecord)) {
                    if ("2".equals(delType)) {
                        lastFilter = new StringBuilder();
                        Map<String, List<Record>> collect = lastNotDeleteRecord.stream().collect(
                                Collectors.groupingBy(item -> StringUtils.isEmpty(item.getSourceInfo()) ? this.collectId : item.getSourceInfo()));
                        for (Map.Entry<String, List<Record>> entry : collect.entrySet()) {
                            if (lastFilter.length() > 0) {
                                lastFilter.append(" OR ");
                            }
                            lastFilter.append(" (");
                            lastFilter.append(BaseUtil.buildQuestionMarkFilter(this.preMasterField,
                                    entry.getValue().stream().map(item -> item.getUniqueValue()[0]).toArray(), true));
                            lastFilter.append(" and ").append(this.sourceInfoField).append("= '").append(entry.getKey()).append("' )");
                        }
                    } else {
                        lastFilter = new StringBuilder(getMultiplePrimaryFieldFilter(lastNotDeleteRecord));
                    }
                }
                StringBuilder sql = new StringBuilder();
                for (int i = 0; i < allTableName.size(); i++) {
                    String tableName = CollectionUtil.get(allTableName, i);
                    if (i > 0) {
                        sql.append(" UNION ALL \n");
                    }
                    sql.append(" SELECT *");
//                            .append(ArrayUtil.join(this.uniqueFields, ","));
                    sql.append(",'").append(tableName).append("' table_name_c_m_t ");
                    if ("2".equals(delType)) {
//                        sql.append(",").append(this.sourceInfoField).append(",").append(preMasterField);
                    }
                    sql.append(" FROM ").append(tableName);
                    sql.append(" WHERE ").append(getMultiplePrimaryFieldFilter(recordList));
                    if (!CollectionUtil.isEmpty(lastNotDeleteRecord)) {
                        sql.append(" OR (");
                        sql.append(lastFilter);
                        sql.append(" )");
                    }
                }
                DataTableEntity dt = targetDao.getList(sql.toString());
                if (DataTableEntity.isEmpty(dt)) {
                    return;
                }
                Map<String, List<Record>> groupTableRecord = new HashMap<>();
                for (int i = 0; i < dt.getRows(); i++) {
                    String tableName = dt.getString(i, "table_name_c_m_t");
                    String[] uniqueValues = new String[this.uniqueFields.length];
                    for (int k = 0; k < this.uniqueFields.length; k++) {
                        uniqueValues[k] = dt.getString(i, this.uniqueFields[k]);
                    }
                    Record record;
                    if ("2".equals(delType)) {
                        record = new Record(new String[]{dt.getString(i, this.preMasterField)}, dt.getString(i, this.sourceInfoField));
                    } else {
                        record = new Record(uniqueValues, null);
                    }
                    int index = recordList.indexOf(record);
                    List<Record> r = groupTableRecord.computeIfAbsent(tableName, k -> new ArrayList<>());
                    if (index != -1) {
                        record = recordList.get(index);
                        if (record.getDeleteFlag() == 2) {
                            String[] primaryValues = new String[this.uniqueFields.length];
                            for (int j = 0; j < this.uniqueFields.length; j++) {
                                primaryValues[j] = dt.getString(i, this.uniqueFields[j]);
                            }
                            record.setPrimaryValue(primaryValues);
                            record.setRecord(dt.getFieldSetEntity(i));
                        }
                        r.add(record);
                    }
                }
                for (Map.Entry<String, List<Record>> entry : groupTableRecord.entrySet()) {
                    if (CollectionUtil.isEmpty(entry.getValue())) {
                        entry.getValue().clear();
                        continue;
                    }
                    String filter = getMultiplePrimaryFieldFilter(entry.getValue());
                    if (StringUtils.isEmpty(filter)) {
                        entry.getValue().clear();
                        continue;
                    }
                    targetDao.delete(entry.getKey(), filter);
                    WriteData(entry.getValue(), dbe.getUuid() + "&_" + entry.getKey() + "&_archiving");
                }
                List<Record> collect = groupTableRecord.values().stream().flatMap(item -> item.stream()).collect(Collectors.toList());
                connection.commit();
                list.removeAll(collect);
            } catch (Exception e) {
                throw e;
            } finally {
                if (connection != null) {
                    connection.close();
                }
            }
        }

        private String getMultiplePrimaryFieldFilter(Collection<Record> recordList) {
            if (CollectionUtil.isEmpty(recordList)) {
                return "1=2";
            }
            StringBuilder sqlFilter = new StringBuilder(128);
            if (this.uniqueFields.length == 1) {
                List<String> uniqueValues = new ArrayList<>();
                for (Record item : recordList) {
//                Record item = CollectionUtil.get(recordList,j);
                    if (ArrayUtil.isEmpty(item.getPrimaryValue())) {
                        continue;
                    }
                    uniqueValues.add(item.getPrimaryValue()[0]);
                }
                if (uniqueValues.size() > 0) {
                    sqlFilter.append(" (").append(BaseUtil.buildQuestionMarkFilter(this.uniqueFields[0], uniqueValues.toArray(), true)).append(" )");
                }
            } else {
                for (Record item : recordList) {
//                Record item = CollectionUtil.get(recordList,j);
                    if (ArrayUtil.isEmpty(item.getPrimaryValue())) {
                        continue;
                    }
                    if (sqlFilter.length() > 0) {
                        sqlFilter.append(" OR ");
                    }
                    sqlFilter.append("( ");
                    for (int k = 0; k < this.uniqueFields.length; k++) {
                        if (k > 0) {
                            sqlFilter.append(" and ");
                        }
                        sqlFilter.append(this.uniqueFields[k]).append("='").append(item.getPrimaryValue()[k]).append("'");
                    }
                    sqlFilter.append(" ) ");
                }
            }
            if (sqlFilter.length() == 0) {
                return "1=2";
            }
            return sqlFilter.toString();
        }


        private StringBuilder getUniqueValues(List<Record> recordList, StringBuilder sql) {
            if (recordList == null || recordList.isEmpty()) {
                return sql;
            }
            if (sql.length() > 0) {
                sql.append("UNION ALL \n");
            }
            sql.append(" SELECT * ");
//                    .append(ArrayUtil.join(ArrayUtil.addAll(this.uniqueFields, new String[]{this.preMasterField, this.sourceInfoField}), ","));
            sql.append(" FROM ").append(this.targetTable);
            String sourceInfo = recordList.get(0).getSourceInfo().toLowerCase(Locale.ROOT);
            sql.append(" WHERE ").append(sourceInfoField).append("='").append(sourceInfo).append("' and (");
            sql.append(BaseUtil.buildQuestionMarkFilter(this.preMasterField, recordList.stream().map(item -> item.getUniqueValue()[0]).toArray(), true)).append(" ) ");
            return sql;
        }

        /**
         * 查询删除记录
         */
        private void selectDelRecord(String uuid) {
            Object o = RedisUtil.get("sync-del-record:" + uuid);
            String filter = "DELETE_FLAG <> 1 and ";
            if (this.sourceDao.getDataBaseType().getValue() == DataBaseType.MYSQL.getValue()) {
                filter += "  CREATE_TIME<date_add(now(),interval -5 minute) and CREATE_TIME>= date_add(DATE_FORMAT(now(),'%Y-%m-%d'),interval -2 DAY)";
            } else {
                filter += " CREATE_TIME< sysdate-((1/24/60)*10)  ";
                if (o != null) {
                    filter += " and CREATE_TIME>=trunc(sysdate-2)";
                } else {
                    filter += " and CREATE_TIME>=trunc(sysdate-30)";
                }
            }
            this.deleteTableRecord = this.sourceDao.getList(this.delRecordTable, filter, null, null);
            for (int i = 0; i < this.deleteTableRecord.getRows(); i++) {
                this.deleteTableRecord.setFieldValue(i, "source_info", StringUtils.lowerCase(this.deleteTableRecord.getString(i, "source_info")));
            }
            RedisUtil.set("sync-del-record:" + uuid, "true");
        }

    }


    /**
     * 创建oracle触发器sql
     *
     * @param tableName    要记录删除的表名
     * @param uniqueFields 记录删除的主键或唯一值字段名
     * @param randomSuffix 随机后缀
     * @param delType      删除类型 0 主库删除 1 采集删除
     * @param collectId    采集id
     * @return
     */
    private String getCreateTriggerSql(String tableName, String[] uniqueFields, String randomSuffix, int delType, String collectField, String collectId) {
        StringBuffer sql = new StringBuffer();
        sql.append(" create or replace trigger \"trigger_").append(randomSuffix).append("\"  after delete on ").append(tableName);
        sql.append(" FOR EACH ROW BEGIN ");
        String idField = "ID_" + randomSuffix.toUpperCase();
        sql.append(" insert into \"trigger_table_").append(randomSuffix).append("\" (" + idField + ",CREATE_TIME,DELETE_FLAG");
        if (2 == delType) {
            sql.append(",SOURCE_INFO, ");
        }
        sql.append(ArrayUtil.join(uniqueFields, ","));
        sql.append(")values(getnewid('trigger_table_");
        sql.append(randomSuffix).append("'),sysdate,0");
        if (2 == delType) {
            sql.append(",nvl(:OLD.").append(collectField.toUpperCase()).append(",'").append(collectId).append("'),");
            sql.append(Arrays.stream(uniqueFields).map(item -> ":OLD." + item.toLowerCase()).collect(Collectors.joining()));
            sql.append(" ");
        }
        sql.append(" );  END; ");
        return sql.toString().toUpperCase();
    }

    /**
     * 获取创建删除记录表sql
     *
     * @param randomSuffix
     * @return
     */
    private String getCreateTableSql(String randomSuffix, int delType, String[] uniqueFields) {
        //修改创建字段请同步修改触发器中的insert sql
        StringBuffer sql = new StringBuffer();
        String idField = "ID_" + randomSuffix.toUpperCase();
        sql.append("CREATE TABLE \"trigger_table_").append(randomSuffix).append("\"");
        sql.append("(	\"" + idField + "\" NUMBER(10,0) NOT NULL ENABLE,\n");
        for (String uniqueField : uniqueFields) {
            sql.append("         \"" + uniqueField.toUpperCase() + "\" VARCHAR2(255) NOT NULL ENABLE,\n");
        }
        if (2 == delType) {
            //采集
            sql.append("       \"SOURCE_INFO\" VARCHAR2(50),\n");
        }
        sql.append("       \"CREATE_TIME\" DATE,\n");
        sql.append("        \"DELETE_FLAG\" NUMBER(2,0),\n");
        sql.append("         PRIMARY KEY (\"" + idField + "\")\n");
        sql.append(" )");
        return sql.toString().toUpperCase();
    }

    /**
     * 获取创建删除记录表sql
     *
     * @param randomSuffix
     * @return
     */
    private String getDropTableSql(String randomSuffix) {
        //修改创建字段请同步修改触发器中的insert sql
        StringBuffer sql = new StringBuffer();
        sql.append("DROP TABLE trigger_table_").append(randomSuffix);
        return sql.toString().toUpperCase();
    }

    /**
     * 创建oracle触发器sql
     *
     * @param randomSuffix 随机后缀
     * @return
     */
    private String getDropTriggerSql(String randomSuffix) {
        StringBuffer sql = new StringBuffer();
        sql.append(" DROP TRIGGER  trigger_").append(randomSuffix);
        return sql.toString().toUpperCase();
    }

    /**
     * 触发器是否存在
     *
     * @param random_suffix
     * @param dao
     * @return
     */
    private boolean triggerIsExist(String random_suffix, Dao dao) {
        FieldSetEntity one = dao.getOne("user_triggers", "TRIGGER_NAME=UPPER(?)", new String[]{"TRIGGER_NAME"}, new Object[]{"trigger_" + random_suffix});
        if (FieldSetEntity.isEmpty(one)) {
            return false;
        }
        return !StringUtils.isEmpty(one.getString("trigger_name"));
    }

    /**
     * 表是否存在
     *
     * @param random_suffix
     * @param dao
     * @return
     */
    private boolean tableIsExist(String random_suffix, Dao dao) {
        FieldSetEntity one = dao.getOne("user_tables", "TABLE_NAME=UPPER(?)", new String[]{"TABLE_NAME"}, new Object[]{"trigger_table_" + random_suffix});
        if (FieldSetEntity.isEmpty(one)) {
            return false;
        }
        return !StringUtils.isEmpty(one.getString("table_name"));
    }

    /**
     * 在来源库，创建删除记录表、触发器
     *
     * @param trigger
     * @return
     */
    public boolean createTriggerDDL(FieldSetEntity trigger) {
        String randomSuffix = trigger.getString("random_suffix");
        if (StringUtils.isEmpty(randomSuffix)) {
            return false;
        }
        DataBaseEntity dbe = new DataBaseEntity(trigger.getString("data_source"));
        Dao dao = dbe.getDao();
        try {
            String[] uniqueFields = trigger.getString("unique_field").split(",");
            Integer delType = trigger.getInteger("del_type");
            if (!tableIsExist(randomSuffix, dao)) {
                dao.executeSql(getCreateTableSql(randomSuffix, delType, uniqueFields));
                //初始化创建自增主键 mes
                dao.executeSql("declare   ids number; begin  ids:=getnewid('trigger_table_" + randomSuffix + "'); dbms_output.put_line(ids);end;");
            }
            if (!triggerIsExist(randomSuffix, dao)) {
                try (Connection conn = dao.getConnection(); Statement statement = conn.createStatement()) {
                    String createTriggerSql = getCreateTriggerSql(trigger.getString("table_name"), uniqueFields, randomSuffix, delType, trigger.getString("collect_field"), trigger.getString("collect_id"));
                    SpringMVCContextHolder.getSystemLogger().info("创建触发器SQL：\n" + createTriggerSql);
                    statement.execute(createTriggerSql);
                }
            }
//            dao.executeSql();
            return true;
        } catch (SQLException e) {
            e.printStackTrace();
            return false;
        } finally {
            dao.closeConnection();
        }
    }

    /**
     * 删除触发器、删除记录表
     *
     * @return
     */
    public boolean deleteTriggerDDL(FieldSetEntity trigger) {
        if (FieldSetEntity.isEmpty(trigger)) {
            return false;
        }
        String randomSuffix = trigger.getString("random_suffix");
        if (StringUtils.isEmpty(randomSuffix)) {
            return false;
        }
        //执行删除trigger ddl语句
        DataBaseEntity dbe = new DataBaseEntity(trigger.getString("data_source"));
        Dao dao = dbe.getDao();
        try {

            if (triggerIsExist(randomSuffix, dao)) {
                dao.executeSql(getDropTriggerSql(randomSuffix));
            }
            if (tableIsExist(randomSuffix, dao)) {
                dao.executeSql(getDropTableSql(randomSuffix));
            }
        } finally {
            dao.closeConnection();
            baseDao.delete(trigger.getTableName(), new String[]{trigger.getUUID()});
        }
        return true;
    }

    /**
     * 删除触发器、删除记录表
     *
     * @param fs 数据采集配置
     * @return
     */
    public boolean deleteTrigger(FieldSetEntity fs) {
        FieldSetEntity trigger = baseDao.getFieldSetByFilter(CmnConst.PRODUCT_MES_DEL_RECORD_CONFIG, "config_uuid=?", new String[]{fs.getUUID()}, false);
        deleteTriggerDDL(trigger);
        return true;
    }

    /**
     * 入口    创建触发器,同步配置调用
     */
    public void createTrigger(DataTableEntity dt, String data_source, String target_data_source) {

        for (int i = 0; i < dt.getRows(); i++) {

            if (CoreConst.SYSTEM_DATA_OPERATE_DEL.equals(dt.getString(i, "~type~")) || !"1".equals(dt.getString(i, "is_used"))) {
                //删除子表数据
                FieldSetEntity trigger = baseDao.getFieldSetByFilter(CmnConst.PRODUCT_MES_DEL_RECORD_CONFIG, "config_uuid=?", new String[]{dt.getFieldSetEntity(i).getUUID()}, false);
                deleteTriggerDDL(trigger);
            } else {
                if (StringUtils.isEmpty(dt.getString(i, CmnConst.UPDATE_TIME))) {
                    //更新时间字段为空的是覆盖对比
                    continue;
                }

                dt.setFieldValue(i, "data_source", data_source);
                dt.setFieldValue(i, "target_data_source", target_data_source);
                dt.setFieldValue(i, "del_type", 1);
                dt.setFieldValue(i, "master_key_field", dt.getString(i, "unique_sign"));
                createTrigger(dt.getFieldSetEntity(i));
            }
        }
    }


    /**
     * 入口    创建触发器，采集配置调用
     *
     * @param fs
     */
    public void createTrigger(FieldSetEntity fs) {
        //采集配置表
//        if (!CmnConst.PRODUCT_SYS_DATA_SYNC_MES_SUB.equals(fs.getTableName())) {
//            fs.setValue("del_type", 2);// 2 子库删除后同步到主库
//            fs.setValue("delete_sync_content", 1);
//        }

        String uuid = "2".equals(fs.getString("del_type")) ? fs.getString("config_uuid") : fs.getUUID();
        FieldSetEntity trigger = null;
//        if (!StringUtils.isEmpty(uuid)) {//新增同步配置
        trigger = baseDao.getFieldSetEntityByFilter(CmnConst.PRODUCT_MES_DEL_RECORD_CONFIG, "config_uuid=?", new String[]{uuid}, false);
        if (FieldSetEntity.isEmpty(trigger)) {
            trigger = null;
        }
//        }
        if ("1".equals(fs.getString("delete_sync_content"))) {//同步删除标识
            if (trigger == null) {
                trigger = new FieldSetEntity();
                trigger.setTableName(CmnConst.PRODUCT_MES_DEL_RECORD_CONFIG);
                trigger.setValue("random_suffix", RandomUtil.randomString(5));

                trigger.setValue("config_uuid", uuid);//同步配置子表数据ID
            }
            trigger.setValue("table_name", fs.getValue("table_name"));
            trigger.setValue("data_source", fs.getValue("data_source"));
            trigger.setValue("del_type", fs.getValue("del_type"));
            trigger.setValue("unique_field", fs.getValue("master_key_field"));//
            trigger.setValue("target_data_source", fs.getValue("target_data_source"));
            //采集表专有的
            trigger.setValue("collect_id", fs.getValue("collect_id"));
            trigger.setValue("collect_field", fs.getValue("collect_field"));
            trigger.setValue("pre_master_field", fs.getValue("pre_master_field"));
            BaseUtil.createCreatorAndCreationTime(trigger);
            baseDao.saveFieldSetEntity(trigger);
            createTriggerDDL(trigger);
        } else if (trigger != null) {//证明，不启动触发器，要删除
            deleteTrigger(fs);
        }
    }
}