DbSyncer/dbsyncer-biz/src/main/java/org/dbsyncer/biz/impl/DataSyncServiceImpl.java

/**
 * DBSyncer Copyright 2020-2024 All Rights Reserved.
 */
package org.dbsyncer.biz.impl;

import com.google.protobuf.InvalidProtocolBufferException;
import org.apache.lucene.index.IndexableField;
import org.dbsyncer.biz.DataSyncService;
import org.dbsyncer.biz.vo.BinlogColumnVo;
import org.dbsyncer.biz.vo.MessageVo;
import org.dbsyncer.common.model.Paging;
import org.dbsyncer.common.util.CollectionUtils;
import org.dbsyncer.common.util.DateFormatUtil;
import org.dbsyncer.common.util.JsonUtil;
import org.dbsyncer.common.util.NumberUtil;
import org.dbsyncer.common.util.StringUtil;
import org.dbsyncer.parser.ProfileComponent;
import org.dbsyncer.parser.flush.impl.BufferActuatorRouter;
import org.dbsyncer.parser.model.Meta;
import org.dbsyncer.parser.model.Picker;
import org.dbsyncer.parser.model.TableGroup;
import org.dbsyncer.sdk.constant.ConfigConstant;
import org.dbsyncer.sdk.constant.ConnectorConstant;
import org.dbsyncer.sdk.enums.StorageEnum;
import org.dbsyncer.sdk.filter.FieldResolver;
import org.dbsyncer.sdk.filter.Query;
import org.dbsyncer.sdk.listener.event.RowChangedEvent;
import org.dbsyncer.sdk.model.Field;
import org.dbsyncer.sdk.storage.StorageService;
import org.dbsyncer.storage.binlog.proto.BinlogMap;
import org.dbsyncer.storage.util.BinlogMessageUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;

import javax.annotation.Resource;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * 数据同步服务
 *
 * @author AE86
 * @version 1.0.0
 * @date 2022/12/19 23:56
 */
@Service
public class DataSyncServiceImpl implements DataSyncService {

    private final Logger logger = LoggerFactory.getLogger(getClass());

    @Resource
    private BufferActuatorRouter bufferActuatorRouter;

    @Resource
    private ProfileComponent profileComponent;

    @Resource
    private StorageService storageService;

    @Override
    public MessageVo getMessageVo(String metaId, String messageId) {
        Assert.hasText(metaId, "The metaId is null.");
        Assert.hasText(messageId, "The messageId is null.");

        MessageVo messageVo = new MessageVo();
        try {
            Map row = getData(metaId, messageId);
            Map binlogData = getBinlogData(row, true);
            String tableGroupId = (String) row.get(ConfigConstant.DATA_TABLE_GROUP_ID);
            TableGroup tableGroup = profileComponent.getTableGroup(tableGroupId);
            messageVo.setSourceTableName(tableGroup.getSourceTable().getName());
            messageVo.setTargetTableName(tableGroup.getTargetTable().getName());
            messageVo.setId(messageId);

            if (!CollectionUtils.isEmpty(binlogData)) {
                Map<String, String> columnMap = tableGroup.getTargetTable().getColumn().stream().collect(Collectors.toMap(Field::getName, Field::getTypeName));
                List<BinlogColumnVo> columns = new ArrayList<>();
                binlogData.forEach((k, v) -> columns.add(new BinlogColumnVo((String) k, v, columnMap.get(k))));
                messageVo.setColumns(columns);
            }
        } catch (Exception e) {
            logger.error(e.getLocalizedMessage());
        }
        return messageVo;
    }

    @Override
    public Map getBinlogData(Map row, boolean prettyBytes) throws InvalidProtocolBufferException {
        String tableGroupId = (String) row.get(ConfigConstant.DATA_TABLE_GROUP_ID);
        // 1、获取配置信息
        final TableGroup tableGroup = profileComponent.getTableGroup(tableGroupId);
        if (tableGroup == null) {
            return Collections.EMPTY_MAP;
        }

        // 2、获取记录的数据
        byte[] bytes = (byte[]) row.get(ConfigConstant.BINLOG_DATA);
        if (null == bytes) {
            if (prettyBytes) {
                String json = (String) row.get(ConfigConstant.CONFIG_MODEL_JSON);
                return JsonUtil.parseMap(json);
            }
            return Collections.EMPTY_MAP;
        }

        // 3、获取DDL
        Map<String, Object> target = new HashMap<>();
        BinlogMap message = BinlogMap.parseFrom(bytes);
        String event = (String) row.get(ConfigConstant.DATA_EVENT);
        if (StringUtil.equals(event, ConnectorConstant.OPERTION_ALTER)) {
            message.getRowMap().forEach((k, v) -> target.put(k, v.toStringUtf8()));
            return target;
        }

        // 4、反序列
        final Picker picker = new Picker(tableGroup);
        final Map<String, Field> fieldMap = picker.getTargetFieldMap();
        message.getRowMap().forEach((k, v) -> {
            if (fieldMap.containsKey(k)) {
                try {
                    Object val = BinlogMessageUtil.deserializeValue(fieldMap.get(k).getType(), v);
                    // 处理二进制对象显示
                    if (prettyBytes) {
                        if (val instanceof byte[]) {
                            byte[] b = (byte[]) val;
                            if (b.length > 128) {
                                target.put(k, String.format("byte[%d]", b.length));
                                return;
                            }
                            target.put(k, Arrays.toString(b));
                            return;
                        }
                    }
                    target.put(k, val);
                } catch (Exception e) {
                    logger.warn("解析Binlog数据类型异常：type=[{}], valueType=[{}], value=[{}]", fieldMap.get(k).getType(),
                            (v == null ? null : v.getClass().getName()), v);
                }
            }
        });
        return target;
    }

    @Override
    public String sync(Map<String, String> params) throws InvalidProtocolBufferException {
        String metaId = params.get("metaId");
        String messageId = params.get("messageId");
        Assert.hasText(metaId, "The metaId is null.");
        Assert.hasText(messageId, "The messageId is null.");

        Map row = getData(metaId, messageId);
        Map binlogData = getBinlogData(row, false);
        if (CollectionUtils.isEmpty(binlogData)) {
            return messageId;
        }
        String tableGroupId = (String) row.get(ConfigConstant.DATA_TABLE_GROUP_ID);
        String event = (String) row.get(ConfigConstant.DATA_EVENT);
        // 有修改同步值
        String retryDataParams = params.get("retryDataParams");
        if (StringUtil.isNotBlank(retryDataParams)) {
            JsonUtil.parseMap(retryDataParams).forEach((k, v) -> binlogData.put(k, convertValue(binlogData.get(k), (String) v)));
        }
        TableGroup tableGroup = profileComponent.getTableGroup(tableGroupId);
        String sourceTableName = tableGroup.getSourceTable().getName();

        // 转换为源字段
        final Picker picker = new Picker(tableGroup);
        List<Object> changedRow = picker.pickSourceData(binlogData);
        RowChangedEvent changedEvent = new RowChangedEvent(sourceTableName, event, changedRow);

        // 执行同步是否成功
        bufferActuatorRouter.execute(metaId, changedEvent);
        storageService.remove(StorageEnum.DATA, metaId, messageId);
        // 更新失败数
        Meta meta = profileComponent.getMeta(metaId);
        Assert.notNull(meta, "Meta can not be null.");
        meta.getFail().decrementAndGet();
        meta.setUpdateTime(Instant.now().toEpochMilli());
        profileComponent.editConfigModel(meta);
        return messageId;
    }

    private Map getData(String metaId, String messageId) {
        Query query = new Query(1, 1);
        Map<String, FieldResolver> fieldResolvers = new ConcurrentHashMap<>();
        fieldResolvers.put(ConfigConstant.BINLOG_DATA, (FieldResolver<IndexableField>) field -> field.binaryValue().bytes);
        query.setFieldResolverMap(fieldResolvers);
        query.addFilter(ConfigConstant.CONFIG_MODEL_ID, messageId);
        query.setMetaId(metaId);
        query.setType(StorageEnum.DATA);
        Paging paging = storageService.query(query);
        if (!CollectionUtils.isEmpty(paging.getData())) {
            List<Map> data = (List<Map>) paging.getData();
            return data.get(0);
        }
        return Collections.EMPTY_MAP;
    }

    private Object convertValue(Object oldValue, String newValue) {
        if (oldValue == null) {
            return newValue;
        }

        Object newVal;
        String type = oldValue.getClass().getName();
        switch (type) {
            case "java.sql.Date":
                newVal = DateFormatUtil.stringToDate(newValue);
                break;
            case "java.sql.Timestamp":
                newVal = DateFormatUtil.stringToTimestamp(newValue);
                break;
            case "java.lang.Integer":
            case "java.lang.Short":
                newVal = NumberUtil.toInt(newValue);
                break;
            case "java.lang.Long":
                newVal = NumberUtil.toLong(newValue);
                break;
            case "java.lang.Float":
                newVal = Float.valueOf(newValue);
                break;
            case "java.lang.Double":
                newVal = Double.valueOf(newValue);
                break;
            case "[B":
                newVal = stringToBytes(newValue);
                break;
            default:
                newVal = newValue;
        }

        return newVal;
    }

    private byte[] stringToBytes(String s) {
        byte[] b = null;
        if (s.startsWith("[") && s.endsWith("]")) {
            s = StringUtil.substring(s, 1, s.length() - 1);
            String[] split = StringUtil.split(s, ",");
            int length = split.length;
            b = new byte[length];
            for (int i = 0; i < length; i++) {
                b[i] = Byte.valueOf(split[i].trim());
            }
        }
        return b;
    }

}