fix: harden financial transaction boundaries

apps/server/src/wallets/wallets.service.ts

@@ -7,6 +7,7 @@ import { StudentWallet } from '../entities/student-wallet.entity';
 import { WalletTransaction } from '../entities/wallet-transaction.entity';
 import { In } from 'typeorm';
 import { BatchChangeWalletBalanceDto, ChangeWalletBalanceDto } from './dto/wallet.dto';
+import { FinancialOperationsService } from '../financial-operations/financial-operations.service';
 
 const money = (value: number | string | null | undefined) => Number(Number(value || 0).toFixed(2));
 
@@ -17,6 +18,7 @@ export class WalletsService {
     @InjectRepository(WalletTransaction) private transactionRepo: Repository<WalletTransaction>,
     @InjectRepository(Student) private studentRepo: Repository<Student>,
     private dataSource: DataSource,
+    private financialOperations?: FinancialOperationsService,
   ) {}
 
   async findAll(query?: { keyword?: string; debtOnly?: boolean }) {
@@ -61,6 +63,18 @@ export class WalletsService {
   }
 
   async changeBalance(dto: ChangeWalletBalanceDto, recordedBy?: number) {
+    const { operationId, ...change } = dto;
+    return this.financialOperations
+      ? this.financialOperations.run(operationId, 'wallet.change_balance', () => this.changeBalanceOnce(change, recordedBy, operationId))
+      : this.changeBalanceOnce(change, recordedBy, operationId);
+  }
+
+  private async changeBalanceOnce(
+    dto: Omit<ChangeWalletBalanceDto, 'operationId'>,
+    recordedBy?: number,
+    operationId?: string,
+    transactionManager?: EntityManager,
+  ) {
     const amount = money(dto.amount);
     if (!Number.isFinite(dto.amount) || Math.abs(dto.amount * 100 - Math.round(dto.amount * 100)) > 1e-8) {
       throw new BadRequestException('调账金额最多保留两位小数');
@@ -69,8 +83,8 @@ export class WalletsService {
     if (dto.type === 'recharge' && amount <= 0) throw new BadRequestException('充值金额必须大于 0');
     const student = await this.studentRepo.findOne({ where: { id: dto.studentId } });
     if (!student) throw new NotFoundException('学生不存在');
-    return this.dataSource.transaction(async (manager) => {
-      const wallet = await this.getOrCreateWallet(manager, dto.studentId);
+    const work = async (manager: EntityManager) => {
+      const wallet = await this.getOrCreateWallet(manager, dto.studentId, true);
       const nextBalance = money(Number(wallet.balance) + amount);
       if (nextBalance < 0) throw new BadRequestException('调账后余额不能小于 0');
       wallet.balance = nextBalance;
@@ -79,6 +93,7 @@ export class WalletsService {
         manager.create(WalletTransaction, {
           studentId: dto.studentId,
           billId: null,
+          operationId: operationId ?? null,
           type: dto.type,
           amount,
           balanceAfter: nextBalance,
@@ -89,21 +104,28 @@ export class WalletsService {
       const payments = amount > 0 ? await this.settleOutstandingBills(manager, dto.studentId, recordedBy) : [];
       const finalWallet = await manager.findOneByOrFail(StudentWallet, { studentId: dto.studentId });
       return { wallet: finalWallet, payments };
-    });
+    };
+    return transactionManager ? work(transactionManager) : this.dataSource.transaction(work);
   }
 
   async batchChangeBalance(dto: BatchChangeWalletBalanceDto, recordedBy?: number) {
-    const uniqueStudentIds = Array.from(new Set(dto.studentIds));
-    const results: Awaited<ReturnType<WalletsService['changeBalance']>>[] = [];
-    for (const studentId of uniqueStudentIds) {
-      results.push(await this.changeBalance({
-        studentId,
-        amount: dto.amount,
-        type: dto.type,
-        description: dto.description,
-      }, recordedBy));
-    }
-    return { count: uniqueStudentIds.length, results };
+    const { operationId, ...batch } = dto;
+    const work = () => this.dataSource.transaction(async (manager) => {
+      const uniqueStudentIds = Array.from(new Set(batch.studentIds));
+      const results: Array<{ wallet: StudentWallet; payments: Bill[] }> = [];
+      for (const studentId of uniqueStudentIds) {
+        results.push(await this.changeBalanceOnce({
+          studentId,
+          amount: batch.amount,
+          type: batch.type,
+          description: batch.description,
+        }, recordedBy, operationId ? `${operationId}:${studentId}` : undefined, manager));
+      }
+      return { count: uniqueStudentIds.length, results };
+    });
+    return this.financialOperations
+      ? this.financialOperations.run(operationId, 'wallet.batch_change_balance', work)
+      : work();
   }
 
   async debitBill(manager: EntityManager, bill: Bill, recordedBy?: number) {
@@ -194,9 +216,27 @@ export class WalletsService {
     return settled;
   }
 
-  private async getOrCreateWallet(manager: EntityManager, studentId: number) {
-    let wallet = await manager.findOne(StudentWallet, { where: { studentId } });
-    if (!wallet) wallet = await manager.save(manager.create(StudentWallet, { studentId, balance: 0 }));
+  private async getOrCreateWallet(manager: EntityManager, studentId: number, lock = false) {
+    const find = async () => {
+      if (!lock || !manager.createQueryBuilder) {
+        return manager.findOne(StudentWallet, { where: { studentId } });
+      }
+      return manager.createQueryBuilder(StudentWallet, 'wallet')
+        .where('wallet.studentId = :studentId', { studentId })
+        .setLock('pessimistic_write')
+        .getOne();
+    };
+    let wallet = await find();
+    if (!wallet) {
+      try {
+        if (manager.insert) await manager.insert(StudentWallet, { studentId, balance: 0 });
+        else wallet = await manager.save(manager.create(StudentWallet, { studentId, balance: 0 }));
+      } catch {
+        // A concurrent request may have inserted the one wallet row.
+      }
+      wallet ||= await find();
+    }
+    if (!wallet) throw new NotFoundException('学生钱包创建失败');
     return wallet;
   }
 }