package com.kxs.transfer.api.controller;
import com.alibaba.fastjson.JSON;
import com.aliyun.dts.subscribe.clients.record.DefaultUserRecord;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.kxs.product.api.model.KxsMerchant;
import com.kxs.system.api.model.KxsSchoolStudy;
import com.kxs.transfer.api.annotation.DtsMsgListener;
import com.kxs.transfer.api.mapper.user.KxsUserByStageMapper;
import com.kxs.transfer.api.model.KxsDtsErrorLog;
import com.kxs.transfer.api.model.KxsDtsLog;
import com.kxs.transfer.api.model.table.DMLData;
import com.kxs.transfer.api.service.KxsDtsErrorLogService;
import com.kxs.transfer.api.service.KxsDtsLogService;
import com.kxs.transfer.api.service.product.*;
import com.kxs.transfer.api.service.store.KxsMachineRecycleService;
import com.kxs.transfer.api.service.store.KxsWarehouseService;
import com.kxs.transfer.api.service.sys.KxsMorningLogService;
import com.kxs.transfer.api.service.sys.KxsMorningService;
import com.kxs.transfer.api.service.sys.KxsSchoolStudyService;
import com.kxs.transfer.api.service.sys.KxsSysMsgService;
import com.kxs.transfer.api.service.user.*;
import com.kxs.user.api.model.KxsPartner;
import com.kxs.user.api.model.KxsUserByStage;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.RestController;
import java.util.List;
/**
*
* DTS控制器
* 在此控制器里做数据分发
* 比如将数据分发到不同的数据源中
*
*
* @author 没秃顶的码农
* @date 2024-01-25
*/
@RestController
@Slf4j
@RequiredArgsConstructor
public class DtsController {
private final KxsDtsLogService kxsDtsLogService;
private final KxsDtsErrorLogService kxsDtsErrorLogService;
private final KxsUserService kxsUserService;
private final KxsUserPresetLogService kxsUserPresetLogService;
private final KxsUserAddressService kxsUserAddressService;
private final KxsUserAmountService kxsUserAmountService;
private final KxsUserAdvanceService kxsUserAdvanceService;
private final KxsUserAmountLogService kxsUserAmountLogService;
private final KxsUserWithdrawalService kxsUserWithdrawalService;
private final KxsLeaderService kxsLeaderService;
private final KxsPartnerService kxsPartnerService;
private final KxsMorningService kxsMorningService;
private final KxsSchoolStudyService kxsSchoolStudyService;
private final KxsSysMsgService kxsSysMsgService;
private final KxsBankChangeLogService kxsBankChangeLogService;
private final KxsUserByStageService kxsUserByStageService;
private final KxsMachineService kxsMachineService;
//产品模块
private final KxsShopGoodsService kxsShopGoodsService;
private final KxsMachinePledgeService kxsMachinePledgeService;
private final KxsMachineRatioService kxsMachineRatioService;
private final KxsTicketService kxsTicketService;
private final KxsTicketTransferService kxsTicketTransferService;
private final KxsMerchantService kxsMerchantService;
//仓库模块
private final KxsWarehouseService kxsWarehouseService;
private final KxsMachineRecycleService kxsMachineRecycleService;
@Value("${spring.profiles.active}")
private String active;
@DtsMsgListener
public void dtsListener(Long dataId, DMLData dmlData, DefaultUserRecord record) {
try {
//过滤无效数据
if ("Users".equals(dmlData.getTableName())) {
List changeFieldList = dmlData.getChangeFieldList();
//过滤Users老表记录登陆设备的信息日志
if (changeFieldList.size() == 2 && changeFieldList.contains("DeviceId") && changeFieldList.contains("DeviceType")) {
record.commit(String.valueOf(record.getSourceTimestamp()));
return;
}
if (changeFieldList.size() == 1 && changeFieldList.contains("DeviceId")) {
// 如果changeFieldList的大小为1,且changeFieldList包含DeviceId,则提交记录
record.commit(String.valueOf(record.getSourceTimestamp()));
return;
}
//过滤Users本月交易额字段
if (changeFieldList.size() == 1 && changeFieldList.contains("ThisMonthTrade")) {
record.commit(String.valueOf(record.getSourceTimestamp()));
return;
}
//过滤Users库存更新字段
if (changeFieldList.size() == 1 && changeFieldList.contains("StoreStock")) {
record.commit(String.valueOf(record.getSourceTimestamp()));
return;
}
}
//去重数据
KxsDtsLog dtsLog = kxsDtsLogService.getOne(Wrappers.lambdaQuery().eq(KxsDtsLog::getDataId, dataId));
if (dtsLog != null) {
log.info("dts的数据重复:{},dataID:{}", dmlData.getValidFieldDataMap(), dataId);
record.commit(String.valueOf(record.getSourceTimestamp()));
return;
}
// if (!"UserMoveInfo".equals(dmlData.getTableName())) {
// log.info("开始消费:{}表的数据,dataID:{}, 表名:{}", dmlData.getTableName(), dataId, dmlData.getDatabaseName());
// }
//用户表
if ("Users".equals(dmlData.getTableName())) {
kxsUserService.changeUser(dmlData);
}
//用户预设职级表
if ("UserRankWhite".equals(dmlData.getTableName())) {
kxsUserPresetLogService.changeUser(dmlData);
}
//密码操作
if ("UserMoveInfo".equals(dmlData.getTableName())) {
kxsUserService.changeUserPwd(dmlData);
//密码操作不存储原始数据
record.commit(String.valueOf(record.getSourceTimestamp()));
return;
}
//用户地址表
if ("UserAddress".equals(dmlData.getTableName())) {
kxsUserAddressService.changeData(dmlData);
}
//用户账户
if ("UserAccount".equals(dmlData.getTableName()) && ("dev".equals(active) || "test".equals(active) ? "KxsMainServer":"KxsProfitServer").equals(dmlData.getDatabaseName())) {
kxsUserAmountService.changeData(dmlData);
}
//用户预扣款表
if ("ToChargeBackRecord".equals(dmlData.getTableName())) {
kxsUserAdvanceService.changeData(dmlData);
}
//用户账户余额日志
if ("UserAccountRecord".equals(dmlData.getTableName())) {
kxsUserAmountLogService.changeData(dmlData);
}
//用户提现申请记录
if ("UserCashRecord".equals(dmlData.getTableName()) && ("dev".equals(active) || "test".equals(active) ? "KxsMainServer":"KxsProfitServer").equals(dmlData.getDatabaseName())) {
kxsUserWithdrawalService.changeData(dmlData);
}
//盟主表
if ("Leaders".equals(dmlData.getTableName())) {
kxsLeaderService.changeData(dmlData);
}
//盟主金额变动记录表
// if ("LeaderReserveRecord".equals(dmlData.getTableName())) {
// kxsLeaderAmountLogService.changeData(dmlData);
// }
//合伙人表
if ("OperatorRankWhite".equals(dmlData.getTableName())) {
kxsPartnerService.changeData(dmlData);
}
//合伙人账户 -> 合伙人表
if ("UserAccount".equals(dmlData.getTableName()) && "KxsOpServer".equals(dmlData.getDatabaseName())) {
kxsPartnerService.changeAccountData(dmlData);
}
//合伙人账户余额变动记录表
if ("AmountRecordNew".equals(dmlData.getTableName()) && "KxsOpServer".equals(dmlData.getDatabaseName())) {
kxsPartnerService.changeAccountLogData(dmlData);
}
//用户结算卡变更记录
if ("SettlementCardChangeRecord".equals(dmlData.getTableName())) {
kxsBankChangeLogService.changeData(dmlData);
}
//分期表数据同步
if ("ToChargeByStage".equals(dmlData.getTableName())) {
kxsUserByStageService.changeData(dmlData);
}
//分期期数表数据同步
if ("ToChargeBackRecordSub".equals(dmlData.getTableName())) {
kxsUserByStageService.changeInfoData(dmlData);
}
//商品表同步
if ("Products".equals(dmlData.getTableName())) {
kxsShopGoodsService.changeData(dmlData);
}
//每日晨会同步
if ("SchoolMorningMeet".equals(dmlData.getTableName())) {
kxsMorningService.changeData(dmlData);
}
if ("SchoolMorningMeetLog".equals(dmlData.getTableName())) {
kxsMorningService.changeLogData(dmlData);
}
//创客学堂
if ("SchoolMakerStudy".equals(dmlData.getTableName())) {
kxsSchoolStudyService.changeData(dmlData);
}
//仓库表
if ("StoreHouse".equals(dmlData.getTableName())) {
kxsWarehouseService.changeData(dmlData);
}
//仓库担保记录
if ("StoreHouseAmountPromiss".equals(dmlData.getTableName())) {
kxsWarehouseService.changeCreditData(dmlData);
}
//仓库额度表
if ("UserAccount".equals(dmlData.getTableName()) && ("dev".equals(active) || "test".equals(active) ? "KxsMainServer":"KxsProfitServer").equals(dmlData.getDatabaseName())) {
kxsWarehouseService.changeAccountData(dmlData);
}
//小仓库额度表
if ("UserAccount".equals(dmlData.getTableName()) && ("dev".equals(active) || "test".equals(active) ? "KxsMainServer":"KxsProfitServer").equals(dmlData.getDatabaseName())) {
kxsWarehouseService.changeAccountMinData(dmlData);
}
//机具表
if ("PosMachinesTwo".equals(dmlData.getTableName())) {
kxsMachineService.changeData(dmlData);
}
//商户表
if ("PosMerchantInfo".equals(dmlData.getTableName())) {
kxsMerchantService.changeData(dmlData);
}
//机具回收表
if ("RecycMachineOrder".equals(dmlData.getTableName())) {
kxsMachineRecycleService.changeData(dmlData);
}
if ("RecycMachineOrderPos".equals(dmlData.getTableName())) {
kxsMachineRecycleService.changeInfoData(dmlData);
}
//机具押金调整记录
if ("MerchantDepositSet".equals(dmlData.getTableName())) {
kxsMachinePledgeService.changeData(dmlData);
}
//机具费率调整记录
if ("PosMachinesFeeChangeRecord".equals(dmlData.getTableName())) {
kxsMachineRatioService.changeData(dmlData);
}
//券码表同步
if ("PosCoupons".equals(dmlData.getTableName())) {
kxsTicketService.changeData(dmlData);
}
//券码划拨表同步
if ("PosCouponOrders".equals(dmlData.getTableName())) {
kxsTicketTransferService.changeData(dmlData);
}
//券码划拨记录表同步
if ("PosCouponRecord".equals(dmlData.getTableName())) {
kxsTicketTransferService.changeInfoData(dmlData);
}
//创客消息表
if ("MsgPersonal".equals(dmlData.getTableName())) {
kxsSysMsgService.changeUserData(dmlData);
}
//系统消息表
if ("MsgPlacard".equals(dmlData.getTableName())) {
kxsSysMsgService.changeSysData(dmlData);
}
if ("MsgPlacardRead".equals(dmlData.getTableName())) {
kxsSysMsgService.changeSysReadData(dmlData);
}
//系统弹窗表
if ("MsgAlert".equals(dmlData.getTableName())) {
kxsSysMsgService.changeModalData(dmlData);
}
//存储历史数据
KxsDtsLog kxsDtsLog = new KxsDtsLog();
kxsDtsLog.setDataId(dataId);
// kxsDtsLog.setContent(JSON.toJSONString(dmlData));
kxsDtsLog.setTableName(dmlData.getTableName());
kxsDtsLog.setOperation(dmlData.getOperation().toString());
kxsDtsLogService.save(kxsDtsLog);
//删除错误记录,不管有没有都删除,因为很快
// kxsDtsErrorLogService.removeById(dataId);
record.commit(String.valueOf(record.getSourceTimestamp()));
} catch (Exception e) {
//详细错误信息
String fullStackTrace = ExceptionUtils.getFullStackTrace(e);
//存储错误数据
KxsDtsErrorLog kxsDtsLog = new KxsDtsErrorLog();
kxsDtsLog.setId(dataId);
kxsDtsLog.setContent(JSON.toJSONString(dmlData));
kxsDtsLog.setTableName(dmlData.getTableName());
kxsDtsLog.setOperation(dmlData.getOperation().toString());
kxsDtsLog.setErrorStr(fullStackTrace);
kxsDtsErrorLogService.save(kxsDtsLog);
record.commit(String.valueOf(record.getSourceTimestamp()));
log.error(fullStackTrace);
}
}
}