package com.kxs.transfer.api.listener;

import com.alibaba.fastjson.JSON;
import com.aliyun.dts.subscribe.clients.common.FieldEntryHolder;
import com.aliyun.dts.subscribe.clients.common.NullableOptional;
import com.aliyun.dts.subscribe.clients.common.RecordListener;
import com.aliyun.dts.subscribe.clients.formats.avro.Field;
import com.aliyun.dts.subscribe.clients.formats.avro.Record;
import com.aliyun.dts.subscribe.clients.record.DefaultUserRecord;
import com.aliyun.dts.subscribe.clients.record.OperationType;
import com.aliyun.dts.subscribe.clients.record.RecordSchema;
import com.kxs.transfer.api.annotation.DtsListener;
import com.kxs.transfer.api.annotation.DtsMsgListener;
import com.kxs.transfer.api.config.DTSProperties;
import com.kxs.transfer.api.converter.MysqlFieldConverter;
import com.kxs.transfer.api.model.table.DMLData;
import com.kxs.transfer.api.model.table.FieldData;
import com.kxs.transfer.api.util.SpringContextUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.*;


/**
 * 消费dts的数据监听器
 *
 * @author ： liukx
 * @time ： 2020/9/22 - 17:22
 */
@Component
@RequiredArgsConstructor
@Slf4j
@DtsListener
@EnableConfigurationProperties(DTSProperties.class)
public class MysqlConsumerListener implements RecordListener {

    private final DTSProperties dtsProperties;


    /**
     * 对应的java属性类型
     */
    private static final Class[] FIELD_CLASS = new Class[256];

    static {

        FIELD_CLASS[0] = BigDecimal.class;

        FIELD_CLASS[1] = Integer.class;
        //Type.INT16;
        FIELD_CLASS[2] = Integer.class;
        //Type.INT32;
        FIELD_CLASS[3] = Integer.class;

        //Type.FLOAT
        FIELD_CLASS[4] = Double.class;
        //Type.DOUBLE
        FIELD_CLASS[5] = Double.class;

        //Type.NULL
        FIELD_CLASS[6] = String.class;

        //Type.TIMESTAMP
        FIELD_CLASS[7] = Date.class;

        //Type.INT64
        FIELD_CLASS[8] = Integer.class;
        //Type.INT24
        FIELD_CLASS[9] = Integer.class;

        //Type.DATE
        FIELD_CLASS[10] = Date.class;
        //Type.TIME
        FIELD_CLASS[11] = Date.class;
        //Type.DATETIME
        FIELD_CLASS[12] = Date.class;
        //Type.YEAR
        FIELD_CLASS[13] = Date.class;
        //Type.DATETIME
        FIELD_CLASS[14] = Date.class;
        //Type.STRING
        FIELD_CLASS[15] = String.class;
        //Type.BIT
        FIELD_CLASS[16] = Integer.class;

        FIELD_CLASS[255] = String.class;    //Type.GEOMETRY;
        FIELD_CLASS[254] = String.class; //Type.STRING;
        FIELD_CLASS[253] = String.class; //Type.STRING;

        FIELD_CLASS[252] = String.class; //Type.BLOB;
        FIELD_CLASS[251] = String.class; //Type.BLOB;
        FIELD_CLASS[250] = String.class; //Type.BLOB;
        FIELD_CLASS[249] = String.class; //Type.BLOB;

        FIELD_CLASS[246] = Long.class; //Type.DECIMAL;

        FIELD_CLASS[248] = String.class; //Type.SET;
        FIELD_CLASS[247] = String.class; //Type.ENUM;
        FIELD_CLASS[245] = String.class;  //Type.JSON;
    }


    @Override
    public void consume(DefaultUserRecord record) {

        OperationType operationType = record.getOperationType();
        if(operationType.equals(OperationType.INSERT)
                || operationType.equals(OperationType.UPDATE)
                || operationType.equals(OperationType.DELETE)
                || operationType.equals(OperationType.DDL)
                || operationType.equals(OperationType.HEARTBEAT)) {

            Record avroRecord = record.getAvroRecord();
            RecordSchema schema = record.getSchema();
            //库名
            NullableOptional<String> databaseName = schema.getDatabaseName();
            NullableOptional<String> tableName = schema.getTableName();
            //过滤空库数据 过滤空表数据
            if (!databaseName.isPresent() || !tableName.isPresent()) {
                record.commit(String.valueOf(record.getSourceTimestamp()));
                return;
            }
            //过滤库和表
            if (!dtsProperties.getFilterDatabase().contains(databaseName.get())
                    || !dtsProperties.getFilterTables().contains(tableName.get())) {
                record.commit(String.valueOf(record.getSourceTimestamp()));
                return;
            }
            //此参数是否唯一过滤数据，还有待考证
            Long dataId = avroRecord.getId();
            DMLData dmlData = parseDML(record);

            SpringContextUtil.invokeMethodByAnnotation(DtsMsgListener.class, dataId, dmlData, record);

        }

    }

    private DMLData parseDML(DefaultUserRecord record) {

        RecordSchema schema = record.getSchema();
        //库名
        NullableOptional<String> databaseName = schema.getDatabaseName();
        //表名
        NullableOptional<String> tableName = schema.getTableName();

        OperationType operationType = record.getOperationType();

        //组装实体数据
        Record avroRecord = record.getAvroRecord();
        DMLData dmlData = new DMLData();
        dmlData.setOperation(operationType);
        dmlData.setDatabaseName(databaseName.get());
        dmlData.setTableName(tableName.get());
        dmlData.setSourceTimestamp(record.getSourceTimestamp());

        List<String> changeFieldList = new ArrayList<>();
        Map<String, FieldData> fieldDataMap = parseUpdateField(avroRecord, changeFieldList);
        if (operationType == OperationType.INSERT || operationType == OperationType.UPDATE) {
            Map<String, Object> validFieldDataMap = parseValidUpdateField(fieldDataMap, changeFieldList);
            dmlData.setValidFieldDataMap(validFieldDataMap);
        }
        dmlData.setId(getFieldValue(fieldDataMap, "Id"));
        dmlData.setFieldDataMap(fieldDataMap);
        dmlData.setChangeFieldList(changeFieldList);
        dmlData.setSourceTimestamp(record.getSourceTimestamp());
        return dmlData;
    }

    /**
     * 解析出有效数据
     *
     * @param fieldDataMap    现场数据映射
     * @param changeFieldList 更改字段列表
     * @return {@link Map}<{@link String}, {@link Object}>
     */
    private Map<String, Object> parseValidUpdateField(Map<String, FieldData> fieldDataMap, List<String> changeFieldList) {
        Map<String, Object> res = new HashMap<>(16);
        for (String f : changeFieldList) {
            FieldData fieldData = fieldDataMap.get(f);
            res.put(f, fieldData.getValue());
        }
        if(fieldDataMap.get("Id") != null){
            res.put("id", fieldDataMap.get("Id").getValue());
        }
        return res;
    }

    /**
     * 获取数据的值
     *
     * @param fieldDataMap 数据对象
     * @param fieldName    数据字段名称
     * @return {@link String}
     */
    private String getFieldValue(Map<String, FieldData> fieldDataMap, String fieldName) {
        FieldData fieldData = fieldDataMap.get(fieldName);
        if (fieldData != null) {
            Object value = fieldData.getValue();
            if (value != null) {
                return value.toString();
            }

            Object oldValue = fieldData.getOldValue();
            if (oldValue != null) {
                return oldValue.toString();
            }
        }
        return null;
    }

    /**
     * 解析更新字段
     * 解析修改类型
     *
     * @param record          记录
     * @param changeFieldList 更改字段列表
     * @return {@link Map}<{@link String}, {@link FieldData}>
     */
    private Map<String, FieldData> parseUpdateField(Record record, List<String> changeFieldList) {
        List<Field> fields = (List<Field>) record.getFields();
        FieldEntryHolder[] fieldArray = getFieldEntryHolder(record);
        FieldEntryHolder before = fieldArray[0];
        FieldEntryHolder after = fieldArray[1];

        Map<String, FieldData> fieldDataMap = new LinkedHashMap<>();
        if (null != fields) {
            Iterator<Field> fieldIterator = fields.iterator();
            while (fieldIterator.hasNext() && before.hasNext() && after.hasNext()) {
                FieldData fieldData = new FieldData();
                Field field = fieldIterator.next();
                fieldData.setField(field.getName());

                Class fieldType = FIELD_CLASS[field.getDataTypeNumber()];
                fieldData.setDataType(fieldType);

                Object toPrintBefore = before.take();
                if (toPrintBefore != null) {
                    Object beforeValue = getTypeValue(field, toPrintBefore);
                    fieldData.setOldValue(beforeValue);
                }

                Object toPrintAfter = after.take();
                if (toPrintAfter != null) {
                    Object afterValue = getTypeValue(field, toPrintAfter);
                    fieldData.setValue(afterValue);
                }

                if (fieldData.getValue() != null) {
                    if (fieldData.getOldValue() != null) {
                        if (!fieldData.getValue().equals(fieldData.getOldValue())) {
                            changeFieldList.add(field.getName());
                        }
                    } else {
                        changeFieldList.add(field.getName());
                    }
                } else if (fieldData.getOldValue() != null) {
                    changeFieldList.add(field.getName());
                }


                fieldDataMap.put(field.getName(), fieldData);
            }
        }
        return fieldDataMap;
    }
    private Object getTypeValue(Field field, Object toPrintBefore) {

        Object text = MysqlFieldConverter.convert(field, toPrintBefore).toString();

        if (text != null && field.getDataTypeNumber() == 7) {
            text = LocalDateTime.parse(text.toString());
        } else if (text != null && field.getDataTypeNumber() == 245) {
            text = JSON.parseObject(text.toString(), Map.class);
        }
        return text;
    }

    private FieldEntryHolder[] getFieldEntryHolder(Record record) {
        // this is a simple impl, may exist unhandled situation
        FieldEntryHolder[] fieldArray = new FieldEntryHolder[2];

        fieldArray[0] = new FieldEntryHolder((List<Object>) record.getBeforeImages());
        fieldArray[1] = new FieldEntryHolder((List<Object>) record.getAfterImages());

        return fieldArray;
    }
}
