rwadurian/backend/services/mining-admin-service/src/infrastructure/kafka/wallet-sync.handlers.ts

751 lines
25 KiB
TypeScript

import { Injectable, Logger } from '@nestjs/common';
import { PrismaService } from '../persistence/prisma/prisma.service';
import { ServiceEvent } from './cdc-consumer.service';
/**
* mining-wallet-service CDC 事件处理器
*/
@Injectable()
export class WalletSyncHandlers {
private readonly logger = new Logger(WalletSyncHandlers.name);
constructor(private readonly prisma: PrismaService) {}
// ===========================================================================
// 区域数据处理
// ===========================================================================
async handleProvinceCreated(event: ServiceEvent): Promise<void> {
const { payload } = event;
try {
await this.prisma.syncedProvince.upsert({
where: { originalId: payload.id },
create: {
originalId: payload.id,
code: payload.code,
name: payload.name,
status: payload.status || 'ACTIVE',
},
update: {
code: payload.code,
name: payload.name,
status: payload.status || 'ACTIVE',
},
});
this.logger.debug(`Synced province: ${payload.code}`);
} catch (error) {
this.logger.error(`Failed to sync province: ${payload.code}`, error);
}
}
async handleProvinceUpdated(event: ServiceEvent): Promise<void> {
const { payload } = event;
try {
await this.prisma.syncedProvince.updateMany({
where: { originalId: payload.id },
data: {
code: payload.code,
name: payload.name,
status: payload.status,
},
});
this.logger.debug(`Updated province: ${payload.code}`);
} catch (error) {
this.logger.error(`Failed to update province: ${payload.code}`, error);
}
}
async handleCityCreated(event: ServiceEvent): Promise<void> {
const { payload } = event;
try {
await this.prisma.syncedCity.upsert({
where: { originalId: payload.id },
create: {
originalId: payload.id,
provinceId: payload.provinceId,
code: payload.code,
name: payload.name,
status: payload.status || 'ACTIVE',
},
update: {
provinceId: payload.provinceId,
code: payload.code,
name: payload.name,
status: payload.status || 'ACTIVE',
},
});
this.logger.debug(`Synced city: ${payload.code}`);
} catch (error) {
this.logger.error(`Failed to sync city: ${payload.code}`, error);
}
}
async handleCityUpdated(event: ServiceEvent): Promise<void> {
const { payload } = event;
try {
await this.prisma.syncedCity.updateMany({
where: { originalId: payload.id },
data: {
provinceId: payload.provinceId,
code: payload.code,
name: payload.name,
status: payload.status,
},
});
this.logger.debug(`Updated city: ${payload.code}`);
} catch (error) {
this.logger.error(`Failed to update city: ${payload.code}`, error);
}
}
async handleUserRegionMappingCreated(event: ServiceEvent): Promise<void> {
const { payload } = event;
try {
await this.prisma.syncedUserRegionMapping.upsert({
where: { accountSequence: payload.accountSequence },
create: {
accountSequence: payload.accountSequence,
cityId: payload.cityId,
assignedAt: new Date(payload.assignedAt),
assignedBy: payload.assignedBy,
},
update: {
cityId: payload.cityId,
assignedAt: new Date(payload.assignedAt),
assignedBy: payload.assignedBy,
},
});
this.logger.debug(`Synced user region mapping: ${payload.accountSequence}`);
} catch (error) {
this.logger.error(`Failed to sync user region mapping: ${payload.accountSequence}`, error);
}
}
async handleUserRegionMappingUpdated(event: ServiceEvent): Promise<void> {
const { payload } = event;
try {
await this.prisma.syncedUserRegionMapping.updateMany({
where: { accountSequence: payload.accountSequence },
data: {
cityId: payload.cityId,
assignedAt: new Date(payload.assignedAt),
assignedBy: payload.assignedBy,
},
});
this.logger.debug(`Updated user region mapping: ${payload.accountSequence}`);
} catch (error) {
this.logger.error(`Failed to update user region mapping: ${payload.accountSequence}`, error);
}
}
// ===========================================================================
// 系统账户处理
// ===========================================================================
async handleWalletSystemAccountCreated(event: ServiceEvent): Promise<void> {
const { payload } = event;
try {
await this.prisma.syncedWalletSystemAccount.upsert({
where: { originalId: payload.id },
create: {
originalId: payload.id,
accountType: payload.accountType,
name: payload.name,
code: payload.code,
provinceId: payload.provinceId,
cityId: payload.cityId,
shareBalance: payload.shareBalance || 0,
usdtBalance: payload.usdtBalance || 0,
greenPointBalance: payload.greenPointBalance || 0,
frozenShare: payload.frozenShare || 0,
frozenUsdt: payload.frozenUsdt || 0,
totalInflow: payload.totalInflow || 0,
totalOutflow: payload.totalOutflow || 0,
blockchainAddress: payload.blockchainAddress,
isActive: payload.isActive ?? true,
},
update: {
accountType: payload.accountType,
name: payload.name,
code: payload.code,
provinceId: payload.provinceId,
cityId: payload.cityId,
shareBalance: payload.shareBalance,
usdtBalance: payload.usdtBalance,
greenPointBalance: payload.greenPointBalance,
frozenShare: payload.frozenShare,
frozenUsdt: payload.frozenUsdt,
totalInflow: payload.totalInflow,
totalOutflow: payload.totalOutflow,
blockchainAddress: payload.blockchainAddress,
isActive: payload.isActive,
},
});
this.logger.debug(`Synced wallet system account: ${payload.code}`);
} catch (error) {
this.logger.error(`Failed to sync wallet system account: ${payload.code}`, error);
}
}
async handleWalletSystemAccountUpdated(event: ServiceEvent): Promise<void> {
const { payload } = event;
try {
await this.prisma.syncedWalletSystemAccount.updateMany({
where: { originalId: payload.id },
data: {
name: payload.name,
shareBalance: payload.shareBalance,
usdtBalance: payload.usdtBalance,
greenPointBalance: payload.greenPointBalance,
frozenShare: payload.frozenShare,
frozenUsdt: payload.frozenUsdt,
totalInflow: payload.totalInflow,
totalOutflow: payload.totalOutflow,
blockchainAddress: payload.blockchainAddress,
isActive: payload.isActive,
},
});
this.logger.debug(`Updated wallet system account: ${payload.code}`);
} catch (error) {
this.logger.error(`Failed to update wallet system account: ${payload.code}`, error);
}
}
// ===========================================================================
// 池账户处理
// ===========================================================================
async handleWalletPoolAccountCreated(event: ServiceEvent): Promise<void> {
const { payload } = event;
try {
await this.prisma.syncedWalletPoolAccount.upsert({
where: { originalId: payload.id },
create: {
originalId: payload.id,
poolType: payload.poolType,
name: payload.name,
balance: payload.balance || 0,
totalInflow: payload.totalInflow || 0,
totalOutflow: payload.totalOutflow || 0,
targetBurn: payload.targetBurn,
remainingBurn: payload.remainingBurn,
isActive: payload.isActive ?? true,
},
update: {
name: payload.name,
balance: payload.balance,
totalInflow: payload.totalInflow,
totalOutflow: payload.totalOutflow,
targetBurn: payload.targetBurn,
remainingBurn: payload.remainingBurn,
isActive: payload.isActive,
},
});
this.logger.debug(`Synced wallet pool account: ${payload.poolType}`);
} catch (error) {
this.logger.error(`Failed to sync wallet pool account: ${payload.poolType}`, error);
}
}
async handleWalletPoolAccountUpdated(event: ServiceEvent): Promise<void> {
const { payload } = event;
try {
await this.prisma.syncedWalletPoolAccount.updateMany({
where: { originalId: payload.id },
data: {
name: payload.name,
balance: payload.balance,
totalInflow: payload.totalInflow,
totalOutflow: payload.totalOutflow,
targetBurn: payload.targetBurn,
remainingBurn: payload.remainingBurn,
isActive: payload.isActive,
},
});
this.logger.debug(`Updated wallet pool account: ${payload.poolType}`);
} catch (error) {
this.logger.error(`Failed to update wallet pool account: ${payload.poolType}`, error);
}
}
// ===========================================================================
// 用户钱包处理
// ===========================================================================
async handleUserWalletCreated(event: ServiceEvent): Promise<void> {
const { payload } = event;
try {
await this.prisma.syncedUserWallet.upsert({
where: { originalId: payload.id },
create: {
originalId: payload.id,
accountSequence: payload.accountSequence,
walletType: payload.walletType,
balance: payload.balance || 0,
frozenBalance: payload.frozenBalance || 0,
totalInflow: payload.totalInflow || 0,
totalOutflow: payload.totalOutflow || 0,
isActive: payload.isActive ?? true,
},
update: {
balance: payload.balance,
frozenBalance: payload.frozenBalance,
totalInflow: payload.totalInflow,
totalOutflow: payload.totalOutflow,
isActive: payload.isActive,
},
});
this.logger.debug(`Synced user wallet: ${payload.accountSequence}/${payload.walletType}`);
} catch (error) {
this.logger.error(`Failed to sync user wallet: ${payload.accountSequence}/${payload.walletType}`, error);
}
}
async handleUserWalletUpdated(event: ServiceEvent): Promise<void> {
const { payload } = event;
try {
await this.prisma.syncedUserWallet.updateMany({
where: { originalId: payload.id },
data: {
balance: payload.balance,
frozenBalance: payload.frozenBalance,
totalInflow: payload.totalInflow,
totalOutflow: payload.totalOutflow,
isActive: payload.isActive,
},
});
this.logger.debug(`Updated user wallet: ${payload.accountSequence}/${payload.walletType}`);
} catch (error) {
this.logger.error(`Failed to update user wallet: ${payload.accountSequence}/${payload.walletType}`, error);
}
}
// ===========================================================================
// 提现请求处理
// ===========================================================================
async handleWithdrawRequestCreated(event: ServiceEvent): Promise<void> {
const { payload } = event;
try {
await this.prisma.syncedWithdrawRequest.upsert({
where: { originalId: payload.id },
create: {
originalId: payload.id,
requestNo: payload.requestNo,
accountSequence: payload.accountSequence,
assetType: payload.assetType,
amount: payload.amount,
fee: payload.fee || 0,
netAmount: payload.netAmount,
toAddress: payload.toAddress,
status: payload.status,
txHash: payload.txHash,
blockNumber: payload.blockNumber,
confirmations: payload.confirmations || 0,
errorMessage: payload.errorMessage,
approvedBy: payload.approvedBy,
approvedAt: payload.approvedAt ? new Date(payload.approvedAt) : null,
createdAt: new Date(payload.createdAt),
completedAt: payload.completedAt ? new Date(payload.completedAt) : null,
},
update: {
status: payload.status,
txHash: payload.txHash,
blockNumber: payload.blockNumber,
confirmations: payload.confirmations,
errorMessage: payload.errorMessage,
approvedBy: payload.approvedBy,
approvedAt: payload.approvedAt ? new Date(payload.approvedAt) : null,
completedAt: payload.completedAt ? new Date(payload.completedAt) : null,
},
});
this.logger.debug(`Synced withdraw request: ${payload.requestNo}`);
} catch (error) {
this.logger.error(`Failed to sync withdraw request: ${payload.requestNo}`, error);
}
}
async handleWithdrawRequestUpdated(event: ServiceEvent): Promise<void> {
const { payload } = event;
try {
await this.prisma.syncedWithdrawRequest.updateMany({
where: { originalId: payload.id },
data: {
status: payload.status,
txHash: payload.txHash,
blockNumber: payload.blockNumber,
confirmations: payload.confirmations,
errorMessage: payload.errorMessage,
approvedBy: payload.approvedBy,
approvedAt: payload.approvedAt ? new Date(payload.approvedAt) : null,
completedAt: payload.completedAt ? new Date(payload.completedAt) : null,
},
});
this.logger.debug(`Updated withdraw request: ${payload.requestNo}`);
} catch (error) {
this.logger.error(`Failed to update withdraw request: ${payload.requestNo}`, error);
}
}
// ===========================================================================
// 充值记录处理
// ===========================================================================
async handleDepositRecordCreated(event: ServiceEvent): Promise<void> {
const { payload } = event;
try {
await this.prisma.syncedDepositRecord.upsert({
where: { originalId: payload.id },
create: {
originalId: payload.id,
txHash: payload.txHash,
fromAddress: payload.fromAddress,
toAddress: payload.toAddress,
assetType: payload.assetType,
amount: payload.amount,
blockNumber: payload.blockNumber,
confirmations: payload.confirmations || 0,
matchedAccountSeq: payload.matchedAccountSeq,
isProcessed: payload.isProcessed || false,
processedAt: payload.processedAt ? new Date(payload.processedAt) : null,
createdAt: new Date(payload.createdAt),
},
update: {
confirmations: payload.confirmations,
matchedAccountSeq: payload.matchedAccountSeq,
isProcessed: payload.isProcessed,
processedAt: payload.processedAt ? new Date(payload.processedAt) : null,
},
});
this.logger.debug(`Synced deposit record: ${payload.txHash}`);
} catch (error) {
this.logger.error(`Failed to sync deposit record: ${payload.txHash}`, error);
}
}
async handleDepositRecordUpdated(event: ServiceEvent): Promise<void> {
const { payload } = event;
try {
await this.prisma.syncedDepositRecord.updateMany({
where: { originalId: payload.id },
data: {
confirmations: payload.confirmations,
matchedAccountSeq: payload.matchedAccountSeq,
isProcessed: payload.isProcessed,
processedAt: payload.processedAt ? new Date(payload.processedAt) : null,
},
});
this.logger.debug(`Updated deposit record: ${payload.txHash}`);
} catch (error) {
this.logger.error(`Failed to update deposit record: ${payload.txHash}`, error);
}
}
// ===========================================================================
// DEX Swap 处理
// ===========================================================================
async handleDexSwapRecordCreated(event: ServiceEvent): Promise<void> {
const { payload } = event;
try {
await this.prisma.syncedDexSwapRecord.upsert({
where: { originalId: payload.id },
create: {
originalId: payload.id,
swapNo: payload.swapNo,
accountSequence: payload.accountSequence,
fromAsset: payload.fromAsset,
toAsset: payload.toAsset,
fromAmount: payload.fromAmount,
toAmount: payload.toAmount,
exchangeRate: payload.exchangeRate,
slippage: payload.slippage || 0,
fee: payload.fee || 0,
status: payload.status,
txHash: payload.txHash,
blockNumber: payload.blockNumber,
errorMessage: payload.errorMessage,
createdAt: new Date(payload.createdAt),
completedAt: payload.completedAt ? new Date(payload.completedAt) : null,
},
update: {
toAmount: payload.toAmount,
exchangeRate: payload.exchangeRate,
status: payload.status,
txHash: payload.txHash,
blockNumber: payload.blockNumber,
errorMessage: payload.errorMessage,
completedAt: payload.completedAt ? new Date(payload.completedAt) : null,
},
});
this.logger.debug(`Synced dex swap record: ${payload.swapNo}`);
} catch (error) {
this.logger.error(`Failed to sync dex swap record: ${payload.swapNo}`, error);
}
}
async handleDexSwapRecordUpdated(event: ServiceEvent): Promise<void> {
const { payload } = event;
try {
await this.prisma.syncedDexSwapRecord.updateMany({
where: { originalId: payload.id },
data: {
toAmount: payload.toAmount,
exchangeRate: payload.exchangeRate,
status: payload.status,
txHash: payload.txHash,
blockNumber: payload.blockNumber,
errorMessage: payload.errorMessage,
completedAt: payload.completedAt ? new Date(payload.completedAt) : null,
},
});
this.logger.debug(`Updated dex swap record: ${payload.swapNo}`);
} catch (error) {
this.logger.error(`Failed to update dex swap record: ${payload.swapNo}`, error);
}
}
// ===========================================================================
// 地址绑定处理
// ===========================================================================
async handleBlockchainAddressBindingCreated(event: ServiceEvent): Promise<void> {
const { payload } = event;
try {
await this.prisma.syncedBlockchainAddressBinding.upsert({
where: { originalId: payload.id },
create: {
originalId: payload.id,
accountSequence: payload.accountSequence,
kavaAddress: payload.kavaAddress,
isVerified: payload.isVerified || false,
verifiedAt: payload.verifiedAt ? new Date(payload.verifiedAt) : null,
verificationTxHash: payload.verificationTxHash,
createdAt: new Date(payload.createdAt),
},
update: {
kavaAddress: payload.kavaAddress,
isVerified: payload.isVerified,
verifiedAt: payload.verifiedAt ? new Date(payload.verifiedAt) : null,
verificationTxHash: payload.verificationTxHash,
},
});
this.logger.debug(`Synced blockchain address binding: ${payload.accountSequence}`);
} catch (error) {
this.logger.error(`Failed to sync blockchain address binding: ${payload.accountSequence}`, error);
}
}
async handleBlockchainAddressBindingUpdated(event: ServiceEvent): Promise<void> {
const { payload } = event;
try {
await this.prisma.syncedBlockchainAddressBinding.updateMany({
where: { originalId: payload.id },
data: {
kavaAddress: payload.kavaAddress,
isVerified: payload.isVerified,
verifiedAt: payload.verifiedAt ? new Date(payload.verifiedAt) : null,
verificationTxHash: payload.verificationTxHash,
},
});
this.logger.debug(`Updated blockchain address binding: ${payload.accountSequence}`);
} catch (error) {
this.logger.error(`Failed to update blockchain address binding: ${payload.accountSequence}`, error);
}
}
// ===========================================================================
// 黑洞合约处理
// ===========================================================================
async handleBlackHoleContractCreated(event: ServiceEvent): Promise<void> {
const { payload } = event;
try {
await this.prisma.syncedBlackHoleContract.upsert({
where: { originalId: payload.id },
create: {
originalId: payload.id,
contractAddress: payload.contractAddress,
name: payload.name,
totalBurned: payload.totalBurned || 0,
targetBurn: payload.targetBurn,
remainingBurn: payload.remainingBurn,
isActive: payload.isActive ?? true,
},
update: {
name: payload.name,
totalBurned: payload.totalBurned,
targetBurn: payload.targetBurn,
remainingBurn: payload.remainingBurn,
isActive: payload.isActive,
},
});
this.logger.debug(`Synced black hole contract: ${payload.contractAddress}`);
} catch (error) {
this.logger.error(`Failed to sync black hole contract: ${payload.contractAddress}`, error);
}
}
async handleBlackHoleContractUpdated(event: ServiceEvent): Promise<void> {
const { payload } = event;
try {
await this.prisma.syncedBlackHoleContract.updateMany({
where: { originalId: payload.id },
data: {
name: payload.name,
totalBurned: payload.totalBurned,
targetBurn: payload.targetBurn,
remainingBurn: payload.remainingBurn,
isActive: payload.isActive,
},
});
this.logger.debug(`Updated black hole contract: ${payload.contractAddress}`);
} catch (error) {
this.logger.error(`Failed to update black hole contract: ${payload.contractAddress}`, error);
}
}
// ===========================================================================
// 销毁记录处理
// ===========================================================================
async handleBurnToBlackHoleRecordCreated(event: ServiceEvent): Promise<void> {
const { payload } = event;
try {
await this.prisma.syncedBurnToBlackHoleRecord.upsert({
where: { originalId: payload.id },
create: {
originalId: payload.id,
blackHoleId: payload.blackHoleId,
amount: payload.amount,
sourceType: payload.sourceType,
sourceAccountSeq: payload.sourceAccountSeq,
sourceUserId: payload.sourceUserId,
sourcePoolType: payload.sourcePoolType,
txHash: payload.txHash,
blockNumber: payload.blockNumber,
memo: payload.memo,
createdAt: new Date(payload.createdAt),
},
update: {
txHash: payload.txHash,
blockNumber: payload.blockNumber,
},
});
this.logger.debug(`Synced burn to black hole record: ${payload.id}`);
} catch (error) {
this.logger.error(`Failed to sync burn to black hole record: ${payload.id}`, error);
}
}
// ===========================================================================
// 费率配置处理
// ===========================================================================
async handleFeeConfigCreated(event: ServiceEvent): Promise<void> {
const { payload } = event;
try {
await this.prisma.syncedFeeConfig.upsert({
where: { originalId: payload.id },
create: {
originalId: payload.id,
feeType: payload.feeType,
feeRate: payload.feeRate,
minFee: payload.minFee,
maxFee: payload.maxFee,
headquartersRate: payload.headquartersRate,
operationRate: payload.operationRate,
provinceRate: payload.provinceRate,
cityRate: payload.cityRate,
isActive: payload.isActive ?? true,
},
update: {
feeRate: payload.feeRate,
minFee: payload.minFee,
maxFee: payload.maxFee,
headquartersRate: payload.headquartersRate,
operationRate: payload.operationRate,
provinceRate: payload.provinceRate,
cityRate: payload.cityRate,
isActive: payload.isActive,
},
});
this.logger.debug(`Synced fee config: ${payload.feeType}`);
} catch (error) {
this.logger.error(`Failed to sync fee config: ${payload.feeType}`, error);
}
}
async handleFeeConfigUpdated(event: ServiceEvent): Promise<void> {
const { payload } = event;
try {
await this.prisma.syncedFeeConfig.updateMany({
where: { originalId: payload.id },
data: {
feeRate: payload.feeRate,
minFee: payload.minFee,
maxFee: payload.maxFee,
headquartersRate: payload.headquartersRate,
operationRate: payload.operationRate,
provinceRate: payload.provinceRate,
cityRate: payload.cityRate,
isActive: payload.isActive,
},
});
this.logger.debug(`Updated fee config: ${payload.feeType}`);
} catch (error) {
this.logger.error(`Failed to update fee config: ${payload.feeType}`, error);
}
}
}