fix: 修复 P1 事务边界问题

- P1-T1 SetTenantGroups/SetGroupModels 删-插包进事务，避免插入失败清空关联
- P1-T2 ApiKeyUpdate 主表与 scope 更新同事务，缓存失效移到提交后
- P1-T3 InvalidateWallet 移到 FulfillOrder 事务提交后，避免缓存污染窗口
- P1-T4 handleDeletionRequest 跨 4 表删改包进事务，杜绝中间态
- P1-T5 subscribePlanTx 加事务级 advisory lock，串行化同租户订阅

internal/logic/admin/model_group.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/gogf/gf/v2/database/gdb"
 	"github.com/gogf/gf/v2/errors/gerror"
 	"github.com/gogf/gf/v2/frame/g"
 	v1 "github.com/qianfree/team-api/api/admin/v1"
@@ -256,27 +257,35 @@ func (s *sAdmin) SetGroupModels(ctx context.Context, req *v1.GroupModelsSetReq)
 		return nil, common.NewBusinessError(consts.CodeModelGroupNotFound, consts.MsgModelGroupNotFound)
 	}
 
-	// 删除旧关联
-	_, err := dao.MdlGroupModels.Ctx(ctx).Where("group_id", req.ID).Delete()
+	// 删-插必须在同一事务内：插入失败（DB 抖动/唯一键冲突）若与删除分离，会清空分组关联，
+	// 影响下游按分组计算的租户可用模型与定价。事务采用 ctx 传播式：闭包内统一 dao.Xxx.Ctx(ctx)。
+	err := g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
+		// 删除旧关联
+		_, err := dao.MdlGroupModels.Ctx(ctx).Where("group_id", req.ID).Delete()
+		if err != nil {
+			return err
+		}
+
+		// 批量插入新关联
+		if len(req.ModelIds) > 0 {
+			insertData := make([]do.MdlGroupModels, 0, len(req.ModelIds))
+			for _, modelID := range req.ModelIds {
+				insertData = append(insertData, do.MdlGroupModels{
+					GroupId: req.ID,
+					ModelId: modelID,
+				})
+			}
+			_, err = dao.MdlGroupModels.Ctx(ctx).Batch(len(insertData)).Insert(insertData)
+			if err != nil {
+				return err
+			}
+		}
+		return nil
+	})
 	if err != nil {
 		return nil, err
 	}
 
-	// 批量插入新关联
-	if len(req.ModelIds) > 0 {
-		insertData := make([]do.MdlGroupModels, 0, len(req.ModelIds))
-		for _, modelID := range req.ModelIds {
-			insertData = append(insertData, do.MdlGroupModels{
-				GroupId: req.ID,
-				ModelId: modelID,
-			})
-		}
-		_, err = dao.MdlGroupModels.Ctx(ctx).Batch(len(insertData)).Insert(insertData)
-		if err != nil {
-			return nil, err
-		}
-	}
-
 	invalidateTenantsInGroup(ctx, req.ID)
 
 	return nil, nil
@@ -321,27 +330,36 @@ func (s *sAdmin) ListTenantGroups(ctx context.Context, req *v1.TenantGroupsListR
 
 // SetTenantGroups 设置租户关联的分组（全量替换）
 func (s *sAdmin) SetTenantGroups(ctx context.Context, req *v1.TenantGroupsSetReq) (*v1.TenantGroupsSetRes, error) {
-	// 删除旧关联
-	_, err := dao.MdlTenantGroups.Ctx(ctx).Where("tenant_id", req.TenantID).Delete()
+	// 删-插必须在同一事务内：插入失败（DB 抖动/唯一键冲突）若与删除分离，会清空租户模型分组，
+	// 影响下游按分组计算的租户可用模型与定价。事务采用 ctx 传播式：闭包内统一 dao.Xxx.Ctx(ctx)，
+	// 参照 tenant/member_model_scope.go 的正确写法。
+	err := g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
+		// 删除旧关联
+		_, err := dao.MdlTenantGroups.Ctx(ctx).Where("tenant_id", req.TenantID).Delete()
+		if err != nil {
+			return err
+		}
+
+		// 批量插入新关联
+		if len(req.GroupIds) > 0 {
+			insertData := make([]do.MdlTenantGroups, 0, len(req.GroupIds))
+			for _, groupID := range req.GroupIds {
+				insertData = append(insertData, do.MdlTenantGroups{
+					TenantId: req.TenantID,
+					GroupId:  groupID,
+				})
+			}
+			_, err = dao.MdlTenantGroups.Ctx(ctx).Batch(len(insertData)).Insert(insertData)
+			if err != nil {
+				return err
+			}
+		}
+		return nil
+	})
 	if err != nil {
 		return nil, err
 	}
 
-	// 批量插入新关联
-	if len(req.GroupIds) > 0 {
-		insertData := make([]do.MdlTenantGroups, 0, len(req.GroupIds))
-		for _, groupID := range req.GroupIds {
-			insertData = append(insertData, do.MdlTenantGroups{
-				TenantId: req.TenantID,
-				GroupId:  groupID,
-			})
-		}
-		_, err = dao.MdlTenantGroups.Ctx(ctx).Batch(len(insertData)).Insert(insertData)
-		if err != nil {
-			return nil, err
-		}
-	}
-
 	invalidateTenantGroupCache(ctx, req.TenantID)
 	billing.ClearTenantPriceCache(ctx, req.TenantID)
 

internal/logic/payment/fulfill.go

@@ -18,7 +18,12 @@ import (
 
 // FulfillOrder 履约订单（事务内完成：履约+更新订单状态）
 func FulfillOrder(ctx context.Context, orderID int64) error {
-	return g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
+	// walletTouched 标记本次履约是否变更了钱包余额；钱包缓存失效必须在事务提交后执行，
+	// 否则并发 GetWallet 可能在提交前读到旧 DB 值并回写 Redis，污染缓存。
+	var walletTouchedTenantID int64
+	var walletTouched bool
+
+	err := g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
 		var order *struct {
 			TenantID    int64   `json:"tenant_id"`
 			UserID      int64   `json:"user_id"`
@@ -65,6 +70,8 @@ func FulfillOrder(ctx context.Context, orderID int64) error {
 			_ = billing.CheckAndUpgradeLevel(ctx, order.TenantID)
 			// 充值后检查是否需要重置低余额预警标记
 			billing.ResetLowBalanceNotified(ctx, order.TenantID)
+			walletTouchedTenantID = order.TenantID
+			walletTouched = true
 
 		default:
 			return gerror.Newf("unsupported order type for fulfillment: %s", order.OrderType)
@@ -90,10 +97,33 @@ func FulfillOrder(ctx context.Context, orderID int64) error {
 		}
 		return nil
 	})
+	if err != nil {
+		return err
+	}
+
+	// 事务提交后再清除钱包两级缓存（进程内 walletCache + Redis），避免充值后 GetWallet 在 300s 内仍返回旧余额。
+	// 放在提交后执行，避免「DEL → 并发 GetWallet 读到旧 DB 值回写」的缓存污染窗口。
+	if walletTouched {
+		billing.InvalidateWallet(ctx, walletTouchedTenantID)
+	}
+	return nil
 }
 
+// planSubscribeLockNS 是套餐订阅串行化 advisory lock 的命名空间前缀。
+// 采用单 key 形式 pg_advisory_xact_lock(bigint)，key = NS + tenantID：
+// NS 取高位常量段，与其它业务模块（直接以主键为 key）的 advisory lock 隔离，避免冲突。
+const planSubscribeLockNS int64 = 9_001_000_000_000_000
+
 // subscribePlanTx 在事务内订阅套餐（依赖调用方传入携带事务的 ctx，内部统一用 dao.Xxx.Ctx(ctx) 传播）
 func subscribePlanTx(ctx context.Context, tenantID int64, planID int64, months int, autoRenew bool) error {
+	// 串行化同一租户的套餐订阅：两个并发履约（兑换码 + 自动续费、双支付回调等）可能同时读到
+	// active 订阅并各自「置 expired + INSERT 新 active」，最终留下两条 active 订阅。
+	// pg_advisory_xact_lock 随事务提交/回滚自动释放，且不占用任何行锁，无死锁风险；
+	// 后到事务阻塞至前者提交（旧 active 已 expired）后再进入临界区，据此跳过重复订阅。
+	if _, err := g.DB().Ctx(ctx).Exec(ctx, "SELECT pg_advisory_xact_lock(?)", planSubscribeLockNS+tenantID); err != nil {
+		return gerror.Wrapf(err, "acquire subscribe advisory lock for tenant %d", tenantID)
+	}
+
 	var plan *struct {
 		MonthlyPrice       float64 `json:"monthly_price"`
 		YearlyPrice        float64 `json:"yearly_price"`
@@ -203,7 +233,8 @@ func creditWalletTx(ctx context.Context, tenantID int64, amount decimal.Decimal,
 		return err
 	}
 
-	// 清除钱包两级缓存（进程内 walletCache + Redis），避免充值后 GetWallet 在 300s 内仍返回旧余额
-	billing.InvalidateWallet(ctx, tenantID)
+	// 注意：钱包缓存失效（billing.InvalidateWallet）不在事务内调用——事务尚未提交时 DEL Redis，
+	// 并发 GetWallet 可能从 DB 读到旧余额并回写 Redis，导致缓存被旧值重新污染。
+	// 失效由调用方 FulfillOrder 在事务提交后统一执行。
 	return nil
 }

internal/logic/task/data_tasks.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/gogf/gf/v2/database/gdb"
 	"github.com/gogf/gf/v2/frame/g"
 
 	"github.com/qianfree/team-api/internal/dao"
@@ -153,31 +154,40 @@ func handleDeletionRequest(ctx context.Context, payload json.RawMessage) (any, e
 
 	LogInfo(ctx, 0, fmt.Sprintf("开始处理租户 %d 的数据删除请求", p.TenantID))
 
-	_, err := dao.TntUsers.Ctx(ctx).
-		Where("tenant_id", p.TenantID).
-		Data(do.TntUsers{
-			DisplayName: "[deleted]",
-			Email:       fmt.Sprintf("deleted_%d@deleted.local", p.TenantID),
-		}).Update()
+	// 跨 4 张表（tnt_users / api_keys / aud_sensitive_access_logs / tnt_tenants）的删除与更新必须
+	// 在同一事务内完成，否则中途失败会留下部分应用的中间态（如 API Key 已禁用但租户未标记 terminated）。
+	// 事务采用 ctx 传播式：闭包内统一 dao.Xxx.Ctx(ctx)，事务由 ctx 自动挂载。
+	err := g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
+		_, err := dao.TntUsers.Ctx(ctx).
+			Where("tenant_id", p.TenantID).
+			Data(do.TntUsers{
+				DisplayName: "[deleted]",
+				Email:       fmt.Sprintf("deleted_%d@deleted.local", p.TenantID),
+			}).Update()
+		if err != nil {
+			LogError(ctx, 0, fmt.Sprintf("匿名化用户数据失败: %v", err))
+			return fmt.Errorf("anonymize users: %w", err)
+		}
+
+		if _, err := dao.ApiKeys.Ctx(ctx).
+			Where("tenant_id", p.TenantID).
+			Data(do.ApiKeys{Status: "disabled"}).Update(); err != nil {
+			return fmt.Errorf("disable api keys: %w", err)
+		}
+
+		if _, err := common.AuditModelCtx(ctx, "aud_sensitive_access_logs").Where("tenant_id", p.TenantID).Delete(); err != nil {
+			return fmt.Errorf("delete sensitive access logs: %w", err)
+		}
+
+		if _, err := dao.TntTenants.Ctx(ctx).
+			Where("id", p.TenantID).
+			Data(do.TntTenants{Status: "terminated"}).Update(); err != nil {
+			return fmt.Errorf("terminate tenant: %w", err)
+		}
+		return nil
+	})
 	if err != nil {
-		LogError(ctx, 0, fmt.Sprintf("匿名化用户数据失败: %v", err))
-		return nil, fmt.Errorf("anonymize users: %w", err)
-	}
-
-	if _, err := dao.ApiKeys.Ctx(ctx).
-		Where("tenant_id", p.TenantID).
-		Data(do.ApiKeys{Status: "disabled"}).Update(); err != nil {
-		return nil, fmt.Errorf("disable api keys: %w", err)
-	}
-
-	if _, err := common.AuditModelCtx(ctx, "aud_sensitive_access_logs").Where("tenant_id", p.TenantID).Delete(); err != nil {
-		return nil, fmt.Errorf("delete sensitive access logs: %w", err)
-	}
-
-	if _, err := dao.TntTenants.Ctx(ctx).
-		Where("id", p.TenantID).
-		Data(do.TntTenants{Status: "terminated"}).Update(); err != nil {
-		return nil, fmt.Errorf("terminate tenant: %w", err)
+		return nil, err
 	}
 
 	proof := map[string]any{

internal/logic/tenant/api_key.go

@@ -443,23 +443,31 @@ func (s *sTenant) ApiKeyUpdate(ctx context.Context, req *v1.TenantApiKeyUpdateRe
 		hasUpdate = true
 	}
 
-	if hasUpdate {
-		_, err = dao.ApiKeys.Ctx(ctx).
-			Where("id", keyID).
-			Where("tenant_id", tenantID).
-			Data(data).
-			Update()
-		if err != nil {
-			return nil, err
+	// 主表更新与模型范围更新必须在同一事务内：前者成功后者失败会留下中间态（如 Key 已改名但
+	// scope 仍为旧值）。外层事务包住两步；updateApiKeyModelScopes 在携带事务的 ctx 下会被
+	// GoFrame 视为嵌套事务（savepoint），与主表更新共享同一事务。缓存失效放在事务提交后执行。
+	err = g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
+		if hasUpdate {
+			_, err = dao.ApiKeys.Ctx(ctx).
+				Where("id", keyID).
+				Where("tenant_id", tenantID).
+				Data(data).
+				Update()
+			if err != nil {
+				return err
+			}
 		}
-	}
 
-	// 更新模型范围（事务内先删后插）
-	if req.ModelNames != nil {
-		err = updateApiKeyModelScopes(ctx, keyID, req.ModelNames)
-		if err != nil {
-			return nil, err
+		// 更新模型范围（事务内先删后插，作为外层事务的 savepoint）
+		if req.ModelNames != nil {
+			if err = updateApiKeyModelScopes(ctx, keyID, req.ModelNames); err != nil {
+				return err
+			}
 		}
+		return nil
+	})
+	if err != nil {
+		return nil, err
 	}
 
 	if hasUpdate || req.ModelNames != nil {

代码分析结果.md

@@ -205,37 +205,42 @@
 
 ### 🔄 P1-Transaction　事务边界类
 
-#### - [ ] P1-T1　SetTenantGroups 删-插无事务
+#### - [x] P1-T1　SetTenantGroups 删-插无事务
 
 - **位置**：`internal/logic/admin/model_group.go:323-349`
 - **问题描述**：先 `Delete()` 旧关联，再 `Batch().Insert()` 新关联，两步不在同一事务。
 - **影响**：插入失败（DB 抖动/唯一键冲突）会清空租户模型分组，影响下游定价。
 - **修复建议**：包 `g.DB().Transaction`，参照项目正确写法 `tenant/member_model_scope.go:88`。
+　**〔已修复〕** `SetTenantGroups` 的 Delete+BatchInsert 包进 `g.DB().Transaction`（ctx 传播式，闭包内统一 `dao.Xxx.Ctx(ctx)`）。同文件 `SetGroupModels` 存在相同删-插分离 bug，一并包进事务。
 
-#### - [ ] P1-T2　ApiKeyUpdate 主表与 scope 更新分离
+#### - [x] P1-T2　ApiKeyUpdate 主表与 scope 更新分离
 
 - **位置**：`internal/logic/tenant/api_key.go:457-463`
 - **问题描述**：Key 主表更新与 `updateApiKeyModelScopes` 两步分离，前者成功后者失败留中间态。
 - **修复建议**：外层事务包住两步。
+　**〔已修复〕** ApiKeyUpdate 主表 `Update` 与 `updateApiKeyModelScopes` 用外层 `g.DB().Transaction` 包住；后者在携带事务的 ctx 下被 GoFrame 视为嵌套事务（savepoint），与主表更新共享同一事务，任一失败整体回滚。`relay.InvalidateApiKey` 缓存失效移到事务提交后执行。
 
-#### - [ ] P1-T3　InvalidateWallet 在事务提交前调用
+#### - [x] P1-T3　InvalidateWallet 在事务提交前调用
 
 - **位置**：`internal/logic/payment/fulfill.go:207`
 - **问题描述**：事务闭包内调 `billing.InvalidateWallet`（Redis DEL），但事务尚未提交。并发 `GetWallet` 可能从 DB 读到旧值回写 Redis。
 - **修复建议**：把 `InvalidateWallet` 移到事务外（`FulfillOrder` 返回后调用）。
+　**〔已修复〕** 移除 `creditWalletTx` 内的 `billing.InvalidateWallet`；`FulfillOrder` 改为先记录 `walletTouched` 标记，事务提交成功后再调 `billing.InvalidateWallet`（与 `redemption.go:152` 的「事务提交后清缓存」模式一致）。
 
-#### - [ ] P1-T4　handleDeletionRequest 跨表无事务
+#### - [x] P1-T4　handleDeletionRequest 跨表无事务
 
 - **位置**：`internal/logic/task/data_tasks.go:111-155`
 - **问题描述**：跨 `tnt_users`、`api_keys`、`aud_sensitive_access_logs`、`tnt_tenants` 4 张表删除/更新无事务。
 - **影响**：中途失败留下部分应用的中间态（如 API Key 已禁用但租户未标记 terminated）。
 - **修复建议**：`g.DB().Transaction` 包裹。
+　**〔已修复〕** 4 张表的匿名化/禁用/删除/终止包进 `g.DB().Transaction`，任一步失败整体回滚，杜绝部分应用的中间态。
 
-#### - [ ] P1-T5　subscribePlanTx 先取消当前订阅无并发保护
+#### - [x] P1-T5　subscribePlanTx 先取消当前订阅无并发保护
 
 - **位置**：`internal/logic/payment/fulfill.go:120-130`
 - **问题描述**：把当前 active 订阅置 expired 再 INSERT 新的，两个并发履约可能都读到 active → 两条 active 订阅。
 - **修复建议**：INSERT 后用唯一约束（tenant_id + status='active'）防重，或事务内 advisory lock。
+　**〔已修复〕** 采用事务内 advisory lock：`subscribePlanTx` 入口 `SELECT pg_advisory_xact_lock(NS+tenantID)`（`planSubscribeLockNS` 高位命名空间段，避免与其他模块冲突）。事务级锁随提交/回滚自动释放，不占用行锁无死锁风险；后到事务阻塞至前者提交（旧 active 已 expired）后进入临界区，串行化同租户订阅。
 
 ---