123 lines
4.7 KiB
TypeScript
123 lines
4.7 KiB
TypeScript
import { Injectable, Logger, OnModuleInit, OnModuleDestroy } from '@nestjs/common';
|
||
import { ConfigService } from '@nestjs/config';
|
||
import { Cron } from '@nestjs/schedule';
|
||
import { Erc20TransferService } from '@/domain/services/erc20-transfer.service';
|
||
import { EvmProviderAdapter } from '@/infrastructure/blockchain/evm-provider.adapter';
|
||
import { ChainType } from '@/domain/value-objects';
|
||
import { ChainTypeEnum } from '@/domain/enums';
|
||
import Redis from 'ioredis';
|
||
|
||
/**
|
||
* 热钱包余额定时更新调度器
|
||
* [2026-01-07] 更新：添加原生代币 (KAVA) 余额缓存
|
||
*
|
||
* 每 5 秒查询热钱包在各链上的余额，并更新到 Redis 缓存：
|
||
* - dUSDT (绿积分) 余额
|
||
* - 原生代币 (KAVA/BNB) 余额
|
||
*
|
||
* wallet-service 在用户发起转账时读取此缓存，预检查热钱包余额是否足够。
|
||
* reporting-service 读取此缓存用于仪表板显示。
|
||
*
|
||
* 注意：使用 Redis DB 0（公共数据库），以便所有服务都能读取。
|
||
*
|
||
* Redis Key 格式:
|
||
* - hot_wallet:dusdt_balance:{chainType} - dUSDT 余额
|
||
* - hot_wallet:native_balance:{chainType} - 原生代币余额 (KAVA/BNB)
|
||
* Redis Value: 余额字符串（如 "10000.00"）
|
||
* TTL: 30 秒（防止服务故障时缓存过期）
|
||
*
|
||
* 回滚方式：恢复此文件到之前的版本（移除原生代币缓存逻辑）
|
||
*/
|
||
@Injectable()
|
||
export class HotWalletBalanceScheduler implements OnModuleInit, OnModuleDestroy {
|
||
private readonly logger = new Logger(HotWalletBalanceScheduler.name);
|
||
|
||
// Redis key 前缀
|
||
private readonly REDIS_KEY_PREFIX_DUSDT = 'hot_wallet:dusdt_balance:';
|
||
private readonly REDIS_KEY_PREFIX_NATIVE = 'hot_wallet:native_balance:';
|
||
|
||
// 缓存过期时间（秒）
|
||
private readonly CACHE_TTL_SECONDS = 30;
|
||
|
||
// 支持的链类型
|
||
private readonly SUPPORTED_CHAINS = [ChainTypeEnum.KAVA, ChainTypeEnum.BSC];
|
||
|
||
// 使用独立的 Redis 连接，连接到 DB 0（公共数据库）
|
||
private readonly sharedRedis: Redis;
|
||
|
||
constructor(
|
||
private readonly configService: ConfigService,
|
||
private readonly transferService: Erc20TransferService,
|
||
private readonly evmProvider: EvmProviderAdapter,
|
||
) {
|
||
// 创建连接到 DB 0 的 Redis 客户端（公共数据库，所有服务可读取）
|
||
this.sharedRedis = new Redis({
|
||
host: this.configService.get<string>('redis.host') || 'localhost',
|
||
port: this.configService.get<number>('redis.port') || 6379,
|
||
password: this.configService.get<string>('redis.password') || undefined,
|
||
db: 0, // 使用 DB 0 作为公共数据库
|
||
});
|
||
|
||
this.sharedRedis.on('connect', () => {
|
||
this.logger.log('[REDIS] Connected to shared Redis DB 0 for hot wallet balance');
|
||
});
|
||
|
||
this.sharedRedis.on('error', (err) => {
|
||
this.logger.error('[REDIS] Shared Redis connection error', err);
|
||
});
|
||
}
|
||
|
||
onModuleDestroy() {
|
||
this.sharedRedis.disconnect();
|
||
}
|
||
|
||
async onModuleInit() {
|
||
this.logger.log('[INIT] HotWalletBalanceScheduler initialized');
|
||
// 启动时立即执行一次
|
||
await this.updateHotWalletBalances();
|
||
}
|
||
|
||
/**
|
||
* 每 5 秒更新热钱包余额到 Redis
|
||
*/
|
||
@Cron('*/5 * * * * *') // 每 5 秒执行
|
||
async updateHotWalletBalances(): Promise<void> {
|
||
for (const chainType of this.SUPPORTED_CHAINS) {
|
||
try {
|
||
// 检查该链是否已配置
|
||
if (!this.transferService.isConfigured(chainType)) {
|
||
this.logger.debug(`[SKIP] Chain ${chainType} not configured, skipping balance update`);
|
||
continue;
|
||
}
|
||
|
||
// 获取热钱包地址
|
||
const hotWalletAddress = this.transferService.getHotWalletAddress(chainType);
|
||
if (!hotWalletAddress) {
|
||
this.logger.debug(`[SKIP] Hot wallet address not configured for ${chainType}`);
|
||
continue;
|
||
}
|
||
|
||
// 查询 dUSDT 余额
|
||
const dusdtBalance = await this.transferService.getHotWalletBalance(chainType);
|
||
const dusdtKey = `${this.REDIS_KEY_PREFIX_DUSDT}${chainType}`;
|
||
await this.sharedRedis.setex(dusdtKey, this.CACHE_TTL_SECONDS, dusdtBalance);
|
||
|
||
// [2026-01-07] 新增：查询原生代币余额 (KAVA/BNB)
|
||
const nativeBalance = await this.evmProvider.getNativeBalance(
|
||
ChainType.fromEnum(chainType),
|
||
hotWalletAddress,
|
||
);
|
||
const nativeKey = `${this.REDIS_KEY_PREFIX_NATIVE}${chainType}`;
|
||
await this.sharedRedis.setex(nativeKey, this.CACHE_TTL_SECONDS, nativeBalance.formatted);
|
||
|
||
this.logger.debug(
|
||
`[UPDATE] ${chainType} hot wallet: dUSDT=${dusdtBalance}, native=${nativeBalance.formatted}`,
|
||
);
|
||
} catch (error) {
|
||
this.logger.error(`[ERROR] Failed to update ${chainType} hot wallet balance`, error);
|
||
// 单链失败不影响其他链的更新
|
||
}
|
||
}
|
||
}
|
||
}
|