feat(wallet-service): 实现 Unit of Work 模式保证 settleToBalance 事务原子性

- 新增 UnitOfWork 接口和实现，使用 Prisma Interactive Transaction - 修改 IWalletAccountRepository 和 ILedgerEntryRepository 接口支持可选事务参数 - 修改仓库实现，支持在事务中执行数据库操作 - 修改 settleToBalance 方法使用 UnitOfWork，确保钱包更新和流水记录原子性 - 注册 UnitOfWorkService 到 InfrastructureModule 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-06 07:50:02 -08:00 · 2026-01-06 07:50:02 -08:00 · bf1c8d2228
parent 7dc25b75d2
commit bf1c8d2228
9 changed files with 160 additions and 47 deletions
--- a/backend/services/wallet-service/src/application/services/wallet-application.service.ts
+++ b/backend/services/wallet-service/src/application/services/wallet-application.service.ts
@ -7,7 +7,8 @@ import {
  IWithdrawalOrderRepository, WITHDRAWAL_ORDER_REPOSITORY,
  IPendingRewardRepository, PENDING_REWARD_REPOSITORY,
 } from '@/domain/repositories';
-import { PrismaService } from '@/infrastructure/persistence/prisma/prisma.service';
+import { PrismaService } from '@/infrastructure/persistence/prisma/prisma.service';
+import { IUnitOfWork, UNIT_OF_WORK } from '@/infrastructure/persistence/unit-of-work';
 import { LedgerEntry, DepositOrder, SettlementOrder, WithdrawalOrder, PendingReward, PendingRewardStatus, WalletAccount } from '@/domain/aggregates';
 import {
  UserId, Money, Hashpower, LedgerEntryType, AssetType, ChainType, SettleCurrency,
@ -96,6 +97,8 @@ export class WalletApplicationService {
    private readonly prisma: PrismaService,
    private readonly feeConfigRepo: FeeConfigRepositoryImpl,
    private readonly identityClient: IdentityClientService,
+    @Inject(UNIT_OF_WORK)
+    private readonly unitOfWork: IUnitOfWork,
  ) {}

  // =============== Commands ===============
@ -922,27 +925,32 @@ export class WalletApplicationService {
      // 2. 生成结算ID
      const settlementId = `STL_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;

-      // 3. 执行钱包结算
-      wallet.settleToBalance(usdtAmount, settlementId);
-      await this.walletRepo.save(wallet);
+      // 3. 在事务中执行钱包结算和记录流水（确保原子性）
+      const savedWallet = await this.unitOfWork.runInTransaction(async (tx) => {
+        // 执行钱包结算
+        wallet.settleToBalance(usdtAmount, settlementId);
+        const updated = await this.walletRepo.save(wallet, { tx });

-      // 4. 记录账本流水（含详细来源信息）
-      const ledgerEntry = LedgerEntry.create({
-        accountSequence: wallet.accountSequence,
-        userId: UserId.create(userId),
-        entryType: LedgerEntryType.REWARD_SETTLED,
-        amount: usdtAmount,
-        balanceAfter: wallet.balances.usdt.available,
-        refOrderId: settlementId,
-        memo: params.memo || `结算 ${params.usdtAmount} 绿积分到钱包余额`,
-        payloadJson: {
-          settlementType: 'SETTLE_TO_BALANCE',
-          rewardEntryIds: params.rewardEntryIds,
-          rewardCount: params.rewardEntryIds.length,
-          breakdown: params.breakdown,
-        },
+        // 记录账本流水（含详细来源信息）
+        const ledgerEntry = LedgerEntry.create({
+          accountSequence: wallet.accountSequence,
+          userId: UserId.create(userId),
+          entryType: LedgerEntryType.REWARD_SETTLED,
+          amount: usdtAmount,
+          balanceAfter: updated.balances.usdt.available,
+          refOrderId: settlementId,
+          memo: params.memo || `结算 ${params.usdtAmount} 绿积分到钱包余额`,
+          payloadJson: {
+            settlementType: 'SETTLE_TO_BALANCE',
+            rewardEntryIds: params.rewardEntryIds,
+            rewardCount: params.rewardEntryIds.length,
+            breakdown: params.breakdown,
+          },
+        });
+        await this.ledgerRepo.save(ledgerEntry, { tx });
+
+        return updated;
      });
-      await this.ledgerRepo.save(ledgerEntry);

      // 5. 使缓存失效
      await this.walletCacheService.invalidateWallet(userId);
@ -953,7 +961,7 @@ export class WalletApplicationService {
        success: true,
        settlementId,
        settledAmount: params.usdtAmount,
-        balanceAfter: wallet.balances.usdt.available.value,
+        balanceAfter: savedWallet.balances.usdt.available.value,
      };
    } catch (error) {
      this.logger.error(`Failed to settle to balance for ${params.accountSequence}: ${error.message}`);
--- a/backend/services/wallet-service/src/domain/repositories/ledger-entry.repository.interface.ts
+++ b/backend/services/wallet-service/src/domain/repositories/ledger-entry.repository.interface.ts
@ -1,5 +1,6 @@
 import { LedgerEntry } from '@/domain/aggregates';
 import { LedgerEntryType, AssetType } from '@/domain/value-objects';
+import { RepositorySaveOptions } from '@/infrastructure/persistence/unit-of-work';

 export interface LedgerFilters {
  entryType?: LedgerEntryType;
@ -22,8 +23,8 @@ export interface PaginatedResult<T> {
 }

 export interface ILedgerEntryRepository {
-  save(entry: LedgerEntry): Promise<LedgerEntry>;
-  saveAll(entries: LedgerEntry[]): Promise<void>;
+  save(entry: LedgerEntry, options?: RepositorySaveOptions): Promise<LedgerEntry>;
+  saveAll(entries: LedgerEntry[], options?: RepositorySaveOptions): Promise<void>;
  findByUserId(userId: bigint, filters?: LedgerFilters, pagination?: Pagination): Promise<PaginatedResult<LedgerEntry>>;
  findByAccountSequence(accountSequence: string, filters?: LedgerFilters, pagination?: Pagination): Promise<PaginatedResult<LedgerEntry>>;
  findByRefOrderId(refOrderId: string): Promise<LedgerEntry[]>;
--- a/backend/services/wallet-service/src/domain/repositories/wallet-account.repository.interface.ts
+++ b/backend/services/wallet-service/src/domain/repositories/wallet-account.repository.interface.ts
@ -1,11 +1,12 @@
 import { WalletAccount } from '@/domain/aggregates';
+import { RepositorySaveOptions } from '@/infrastructure/persistence/unit-of-work';

 export interface IWalletAccountRepository {
-  save(wallet: WalletAccount): Promise<WalletAccount>;
-  findById(walletId: bigint): Promise<WalletAccount | null>;
-  findByUserId(userId: bigint): Promise<WalletAccount | null>;
-  findByAccountSequence(accountSequence: string): Promise<WalletAccount | null>;
-  getOrCreate(accountSequence: string, userId: bigint): Promise<WalletAccount>;
+  save(wallet: WalletAccount, options?: RepositorySaveOptions): Promise<WalletAccount>;
+  findById(walletId: bigint, options?: RepositorySaveOptions): Promise<WalletAccount | null>;
+  findByUserId(userId: bigint, options?: RepositorySaveOptions): Promise<WalletAccount | null>;
+  findByAccountSequence(accountSequence: string, options?: RepositorySaveOptions): Promise<WalletAccount | null>;
+  getOrCreate(accountSequence: string, userId: bigint, options?: RepositorySaveOptions): Promise<WalletAccount>;
  findByUserIds(userIds: bigint[]): Promise<Map<string, WalletAccount>>;
 }

--- a/backend/services/wallet-service/src/infrastructure/infrastructure.module.ts
+++ b/backend/services/wallet-service/src/infrastructure/infrastructure.module.ts
@ -10,6 +10,7 @@ import {
  PendingRewardRepositoryImpl,
  FeeConfigRepositoryImpl,
 } from './persistence/repositories';
+import { UnitOfWorkService, UNIT_OF_WORK } from './persistence/unit-of-work';
 import {
  WALLET_ACCOUNT_REPOSITORY,
  LEDGER_ENTRY_REPOSITORY,
@ -55,10 +56,15 @@ const repositories = [
  FeeConfigRepositoryImpl,
 ];

+const unitOfWork = {
+  provide: UNIT_OF_WORK,
+  useClass: UnitOfWorkService,
+};
+
@Global()
@Module({
  imports: [RedisModule, KafkaModule, IdentityModule],
-  providers: [PrismaService, ...repositories],
-  exports: [PrismaService, RedisModule, KafkaModule, IdentityModule, FeeConfigRepositoryImpl, ...repositories],
+  providers: [PrismaService, unitOfWork, ...repositories],
+  exports: [PrismaService, unitOfWork, RedisModule, KafkaModule, IdentityModule, FeeConfigRepositoryImpl, ...repositories],
 })
 export class InfrastructureModule {}
--- a/backend/services/wallet-service/src/infrastructure/persistence/repositories/ledger-entry.repository.impl.ts
+++ b/backend/services/wallet-service/src/infrastructure/persistence/repositories/ledger-entry.repository.impl.ts
@ -5,14 +5,20 @@ import {
  ILedgerEntryRepository, LedgerFilters, Pagination, PaginatedResult,
 } from '@/domain/repositories';
 import { LedgerEntry } from '@/domain/aggregates';
+import { RepositorySaveOptions, PrismaTransactionClient } from '@/infrastructure/persistence/unit-of-work';
 import Decimal from 'decimal.js';

@Injectable()
 export class LedgerEntryRepositoryImpl implements ILedgerEntryRepository {
  constructor(private readonly prisma: PrismaService) {}

-  async save(entry: LedgerEntry): Promise<LedgerEntry> {
-    const created = await this.prisma.ledgerEntry.create({
+  private getClient(options?: RepositorySaveOptions): PrismaService | PrismaTransactionClient {
+    return options?.tx || this.prisma;
+  }
+
+  async save(entry: LedgerEntry, options?: RepositorySaveOptions): Promise<LedgerEntry> {
+    const client = this.getClient(options);
+    const created = await client.ledgerEntry.create({
      data: {
        accountSequence: entry.accountSequence,
        userId: entry.userId.value,
@ -29,8 +35,9 @@ export class LedgerEntryRepositoryImpl implements ILedgerEntryRepository {
    return this.toDomain(created);
  }

-  async saveAll(entries: LedgerEntry[]): Promise<void> {
-    await this.prisma.ledgerEntry.createMany({
+  async saveAll(entries: LedgerEntry[], options?: RepositorySaveOptions): Promise<void> {
+    const client = this.getClient(options);
+    await client.ledgerEntry.createMany({
      data: entries.map(entry => ({
        accountSequence: entry.accountSequence,
        userId: entry.userId.value,
--- a/backend/services/wallet-service/src/infrastructure/persistence/repositories/wallet-account.repository.impl.ts
+++ b/backend/services/wallet-service/src/infrastructure/persistence/repositories/wallet-account.repository.impl.ts
@ -4,6 +4,7 @@ import { IWalletAccountRepository } from '@/domain/repositories';
 import { WalletAccount } from '@/domain/aggregates';
 import { UserId, WalletStatus } from '@/domain/value-objects';
 import { OptimisticLockError } from '@/shared/exceptions/domain.exception';
+import { RepositorySaveOptions, PrismaTransactionClient } from '@/infrastructure/persistence/unit-of-work';
 import Decimal from 'decimal.js';

@Injectable()
@ -12,7 +13,12 @@ export class WalletAccountRepositoryImpl implements IWalletAccountRepository {

  constructor(private readonly prisma: PrismaService) {}

-  async save(wallet: WalletAccount): Promise<WalletAccount> {
+  private getClient(options?: RepositorySaveOptions): PrismaService | PrismaTransactionClient {
+    return options?.tx || this.prisma;
+  }
+
+  async save(wallet: WalletAccount, options?: RepositorySaveOptions): Promise<WalletAccount> {
+    const client = this.getClient(options);
    const data = {
      accountSequence: wallet.accountSequence,
      userId: wallet.userId.value,
@ -42,7 +48,7 @@ export class WalletAccountRepositoryImpl implements IWalletAccountRepository {

    if (wallet.walletId.value === BigInt(0)) {
      // Create new wallet with version = 0
-      const created = await this.prisma.walletAccount.create({
+      const created = await client.walletAccount.create({
        data: {
          ...data,
          version: 0,
@ -54,7 +60,7 @@ export class WalletAccountRepositoryImpl implements IWalletAccountRepository {
      const currentVersion = wallet.version;
      const newVersion = currentVersion + 1;

-      const result = await this.prisma.walletAccount.updateMany({
+      const result = await client.walletAccount.updateMany({
        where: {
          id: wallet.walletId.value,
          version: currentVersion, // Optimistic lock: only update if version matches
@ -81,7 +87,7 @@ export class WalletAccountRepositoryImpl implements IWalletAccountRepository {
      );

      // Fetch the updated record to return
-      const updated = await this.prisma.walletAccount.findUnique({
+      const updated = await client.walletAccount.findUnique({
        where: { id: wallet.walletId.value },
      });

@ -93,35 +99,38 @@ export class WalletAccountRepositoryImpl implements IWalletAccountRepository {
    }
  }

-  async findById(walletId: bigint): Promise<WalletAccount | null> {
-    const record = await this.prisma.walletAccount.findUnique({
+  async findById(walletId: bigint, options?: RepositorySaveOptions): Promise<WalletAccount | null> {
+    const client = this.getClient(options);
+    const record = await client.walletAccount.findUnique({
      where: { id: walletId },
    });
    return record ? this.toDomain(record) : null;
  }

-  async findByUserId(userId: bigint): Promise<WalletAccount | null> {
-    const record = await this.prisma.walletAccount.findUnique({
+  async findByUserId(userId: bigint, options?: RepositorySaveOptions): Promise<WalletAccount | null> {
+    const client = this.getClient(options);
+    const record = await client.walletAccount.findUnique({
      where: { userId },
    });
    return record ? this.toDomain(record) : null;
  }

-  async findByAccountSequence(accountSequence: string): Promise<WalletAccount | null> {
-    const record = await this.prisma.walletAccount.findUnique({
+  async findByAccountSequence(accountSequence: string, options?: RepositorySaveOptions): Promise<WalletAccount | null> {
+    const client = this.getClient(options);
+    const record = await client.walletAccount.findUnique({
      where: { accountSequence },
    });
    return record ? this.toDomain(record) : null;
  }

-  async getOrCreate(accountSequence: string, userId: bigint): Promise<WalletAccount> {
-    const existing = await this.findByAccountSequence(accountSequence);
+  async getOrCreate(accountSequence: string, userId: bigint, options?: RepositorySaveOptions): Promise<WalletAccount> {
+    const existing = await this.findByAccountSequence(accountSequence, options);
    if (existing) {
      return existing;
    }

    const newWallet = WalletAccount.createNew(accountSequence, UserId.create(userId));
-    return this.save(newWallet);
+    return this.save(newWallet, options);
  }

  async findByUserIds(userIds: bigint[]): Promise<Map<string, WalletAccount>> {
--- a/backend/services/wallet-service/src/infrastructure/persistence/unit-of-work/index.ts
+++ b/backend/services/wallet-service/src/infrastructure/persistence/unit-of-work/index.ts
@ -0,0 +1,2 @@
+export * from './unit-of-work.interface';
+export * from './unit-of-work.service';
--- a/backend/services/wallet-service/src/infrastructure/persistence/unit-of-work/unit-of-work.interface.ts
+++ b/backend/services/wallet-service/src/infrastructure/persistence/unit-of-work/unit-of-work.interface.ts
@ -0,0 +1,35 @@
+import { PrismaClient } from '@prisma/client';
+
+/**
+ * Prisma 事务客户端类型
+ * 用于在事务中传递给仓库
+ */
+export type PrismaTransactionClient = Omit<
+  PrismaClient,
+  '$connect' | '$disconnect' | '$on' | '$transaction' | '$use' | '$extends'
+>;
+
+/**
+ * 仓库保存选项
+ * 允许传入事务客户端以在 Unit of Work 中使用
+ */
+export interface RepositorySaveOptions {
+  tx?: PrismaTransactionClient;
+}
+
+/**
+ * Unit of Work 接口
+ * 用于管理跨多个仓库的事务
+ */
+export interface IUnitOfWork {
+  /**
+   * 在事务中执行工作
+   * @param work 要执行的工作函数，接收事务客户端
+   * @returns 工作函数的返回值
+   */
+  runInTransaction<T>(
+    work: (tx: PrismaTransactionClient) => Promise<T>,
+  ): Promise<T>;
+}
+
+export const UNIT_OF_WORK = Symbol('IUnitOfWork');
--- a/backend/services/wallet-service/src/infrastructure/persistence/unit-of-work/unit-of-work.service.ts
+++ b/backend/services/wallet-service/src/infrastructure/persistence/unit-of-work/unit-of-work.service.ts
@ -0,0 +1,44 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { PrismaService } from '../prisma/prisma.service';
+import { IUnitOfWork, PrismaTransactionClient } from './unit-of-work.interface';
+
+/**
+ * Unit of Work 实现
+ * 使用 Prisma Interactive Transaction 确保多个仓库操作的原子性
+ */
+@Injectable()
+export class UnitOfWorkService implements IUnitOfWork {
+  private readonly logger = new Logger(UnitOfWorkService.name);
+
+  constructor(private readonly prisma: PrismaService) {}
+
+  /**
+   * 在事务中执行工作
+   * 事务中的所有操作要么全部成功，要么全部回滚
+   */
+  async runInTransaction<T>(
+    work: (tx: PrismaTransactionClient) => Promise<T>,
+  ): Promise<T> {
+    this.logger.debug('[UoW] Starting transaction');
+
+    try {
+      const result = await this.prisma.$transaction(
+        async (tx) => {
+          return await work(tx);
+        },
+        {
+          // 设置较长的超时时间，因为某些操作可能需要更多时间
+          timeout: 30000, // 30 秒
+          // 使用 ReadCommitted 隔离级别，平衡一致性和性能
+          isolationLevel: 'ReadCommitted',
+        },
+      );
+
+      this.logger.debug('[UoW] Transaction committed successfully');
+      return result;
+    } catch (error) {
+      this.logger.error('[UoW] Transaction rolled back due to error', error);
+      throw error;
+    }
+  }
+}