From f354aa7d88331b41bb492643963bdea88d453b85 Mon Sep 17 00:00:00 2001 From: zhx Date: Wed, 22 Jul 2026 15:55:17 +0800 Subject: [PATCH] =?UTF-8?q?=E7=9B=B8=E5=85=B3bug=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- api/proto/wallet/v1/wallet.pb.go | 41 +++- api/proto/wallet/v1/wallet.proto | 5 + scripts/apply-local-mysql-initdb.sh | 1 + server/admin/internal/model/models.go | 2 +- .../modules/hostagencypolicy/service.go | 109 ++++++++-- .../modules/hostagencypolicy/service_test.go | 65 +++++- .../modules/hostsalarysettlement/service.go | 8 +- .../host_agency_salary_policy_repository.go | 21 +- .../transport/http/activityapi/handler.go | 12 +- .../activityapi/user_leaderboard_handler.go | 78 +++++-- .../user_leaderboard_handler_test.go | 52 ++++- .../mysql/initdb/001_wallet_service.sql | 11 +- .../018_host_salary_policy_revisions.sql | 64 ++++++ .../internal/domain/resource/entitlement.go | 12 +- .../vip_user_settings_integration_test.go | 17 ++ .../storage/mysql/host_salary_settlement.go | 8 +- .../mysql/resource_entitlement_repository.go | 14 +- .../internal/storage/mysql/schema.go | 44 +++- .../mysql/vip_benefit_resource_repository.go | 49 +---- .../storage/mysql/vip_program_repository.go | 194 +++++++++++++++++- .../internal/testutil/mysqltest/mysqltest.go | 8 +- .../internal/transport/grpc/resource.go | 12 +- tools/databi-real-flow/main.go | 6 +- 23 files changed, 691 insertions(+), 142 deletions(-) create mode 100644 services/wallet-service/deploy/mysql/migrations/018_host_salary_policy_revisions.sql diff --git a/api/proto/wallet/v1/wallet.pb.go b/api/proto/wallet/v1/wallet.pb.go index e00cb1f8..55666896 100644 --- a/api/proto/wallet/v1/wallet.pb.go +++ b/api/proto/wallet/v1/wallet.pb.go @@ -10212,8 +10212,11 @@ type BatchGetUserEquippedResourcesRequest struct { AppCode string `protobuf:"bytes,2,opt,name=app_code,json=appCode,proto3" json:"app_code,omitempty"` UserIds []int64 `protobuf:"varint,3,rep,packed,name=user_ids,json=userIds,proto3" json:"user_ids,omitempty"` ResourceTypes []string `protobuf:"bytes,4,rep,name=resource_types,json=resourceTypes,proto3" json:"resource_types,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // vip_benefit_codes 只在调用方需要批量装配展示权限时传入；wallet 会同时校验 + // 当前 effective VIP、精确等级权益、体验卡可用性和用户开关，避免调用方逐用户 N+1 校验。 + VipBenefitCodes []string `protobuf:"bytes,5,rep,name=vip_benefit_codes,json=vipBenefitCodes,proto3" json:"vip_benefit_codes,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *BatchGetUserEquippedResourcesRequest) Reset() { @@ -10274,12 +10277,21 @@ func (x *BatchGetUserEquippedResourcesRequest) GetResourceTypes() []string { return nil } +func (x *BatchGetUserEquippedResourcesRequest) GetVipBenefitCodes() []string { + if x != nil { + return x.VipBenefitCodes + } + return nil +} + type UserEquippedResources struct { - state protoimpl.MessageState `protogen:"open.v1"` - UserId int64 `protobuf:"varint,1,opt,name=user_id,json=userId,proto3" json:"user_id,omitempty"` - Resources []*UserResourceEntitlement `protobuf:"bytes,2,rep,name=resources,proto3" json:"resources,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"open.v1"` + UserId int64 `protobuf:"varint,1,opt,name=user_id,json=userId,proto3" json:"user_id,omitempty"` + Resources []*UserResourceEntitlement `protobuf:"bytes,2,rep,name=resources,proto3" json:"resources,omitempty"` + // allowed_vip_benefit_codes 只返回请求中最终可执行的权益；设置值本身不能越权进入此集合。 + AllowedVipBenefitCodes []string `protobuf:"bytes,3,rep,name=allowed_vip_benefit_codes,json=allowedVipBenefitCodes,proto3" json:"allowed_vip_benefit_codes,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *UserEquippedResources) Reset() { @@ -10326,6 +10338,13 @@ func (x *UserEquippedResources) GetResources() []*UserResourceEntitlement { return nil } +func (x *UserEquippedResources) GetAllowedVipBenefitCodes() []string { + if x != nil { + return x.AllowedVipBenefitCodes + } + return nil +} + type BatchGetUserEquippedResourcesResponse struct { state protoimpl.MessageState `protogen:"open.v1"` Users []*UserEquippedResources `protobuf:"bytes,1,rep,name=users,proto3" json:"users,omitempty"` @@ -28199,16 +28218,18 @@ const file_proto_wallet_v1_wallet_proto_rawDesc = "" + "\n" + "unequipped\x18\x02 \x01(\bR\n" + "unequipped\x12\"\n" + - "\rupdated_at_ms\x18\x03 \x01(\x03R\vupdatedAtMs\"\xa2\x01\n" + + "\rupdated_at_ms\x18\x03 \x01(\x03R\vupdatedAtMs\"\xce\x01\n" + "$BatchGetUserEquippedResourcesRequest\x12\x1d\n" + "\n" + "request_id\x18\x01 \x01(\tR\trequestId\x12\x19\n" + "\bapp_code\x18\x02 \x01(\tR\aappCode\x12\x19\n" + "\buser_ids\x18\x03 \x03(\x03R\auserIds\x12%\n" + - "\x0eresource_types\x18\x04 \x03(\tR\rresourceTypes\"x\n" + + "\x0eresource_types\x18\x04 \x03(\tR\rresourceTypes\x12*\n" + + "\x11vip_benefit_codes\x18\x05 \x03(\tR\x0fvipBenefitCodes\"\xb3\x01\n" + "\x15UserEquippedResources\x12\x17\n" + "\auser_id\x18\x01 \x01(\x03R\x06userId\x12F\n" + - "\tresources\x18\x02 \x03(\v2(.hyapp.wallet.v1.UserResourceEntitlementR\tresources\"e\n" + + "\tresources\x18\x02 \x03(\v2(.hyapp.wallet.v1.UserResourceEntitlementR\tresources\x129\n" + + "\x19allowed_vip_benefit_codes\x18\x03 \x03(\tR\x16allowedVipBenefitCodes\"e\n" + "%BatchGetUserEquippedResourcesResponse\x12<\n" + "\x05users\x18\x01 \x03(\v2&.hyapp.wallet.v1.UserEquippedResourcesR\x05users\"\xc4\x01\n" + "\x19ListResourceGrantsRequest\x12\x1d\n" + diff --git a/api/proto/wallet/v1/wallet.proto b/api/proto/wallet/v1/wallet.proto index 9fc92682..3c5be45c 100644 --- a/api/proto/wallet/v1/wallet.proto +++ b/api/proto/wallet/v1/wallet.proto @@ -1207,11 +1207,16 @@ message BatchGetUserEquippedResourcesRequest { string app_code = 2; repeated int64 user_ids = 3; repeated string resource_types = 4; + // vip_benefit_codes 只在调用方需要批量装配展示权限时传入；wallet 会同时校验 + // 当前 effective VIP、精确等级权益、体验卡可用性和用户开关，避免调用方逐用户 N+1 校验。 + repeated string vip_benefit_codes = 5; } message UserEquippedResources { int64 user_id = 1; repeated UserResourceEntitlement resources = 2; + // allowed_vip_benefit_codes 只返回请求中最终可执行的权益；设置值本身不能越权进入此集合。 + repeated string allowed_vip_benefit_codes = 3; } message BatchGetUserEquippedResourcesResponse { diff --git a/scripts/apply-local-mysql-initdb.sh b/scripts/apply-local-mysql-initdb.sh index abae5caa..989d6f5f 100755 --- a/scripts/apply-local-mysql-initdb.sh +++ b/scripts/apply-local-mysql-initdb.sh @@ -47,6 +47,7 @@ INCREMENTAL_MIGRATION_FILES=( "services/wallet-service/deploy/mysql/migrations/011_vip_user_settings_defaults.sql" "services/wallet-service/deploy/mysql/migrations/012_wallet_outbox_archive_receipts.sql" "services/wallet-service/deploy/mysql/migrations/013_wallet_outbox_archive_purge_receipts.sql" + "services/wallet-service/deploy/mysql/migrations/018_host_salary_policy_revisions.sql" "services/lucky-gift-service/deploy/mysql/migrations/013_jackpot_token_version_rate_limit.sql" ) diff --git a/server/admin/internal/model/models.go b/server/admin/internal/model/models.go index e402dfa2..26405ccc 100644 --- a/server/admin/internal/model/models.go +++ b/server/admin/internal/model/models.go @@ -436,7 +436,7 @@ type HostAgencySalaryPolicy struct { RegionID int64 `gorm:"uniqueIndex:uk_admin_host_agency_salary_policy_name;uniqueIndex:uk_admin_host_agency_salary_policy_version;index:idx_admin_host_agency_salary_policy_region;index:idx_admin_host_agency_salary_policy_cycle,not null" json:"regionId"` // RegionIDs 是这份政策实际覆盖的区域集合；RegionID 只保留首个值用于旧客户端和旧索引兼容。 RegionIDs []int64 `gorm:"-" json:"regionIds"` - // CycleKey 是工资归属月；PolicyVersion 使用政策主键形成稳定版本号，发布后两者均不可修改。 + // CycleKey 是工资归属月；PolicyVersion 首次发布等于主键，当前/未来月份修订时递增并对应一份不可变 wallet 快照。 CycleKey string `gorm:"column:cycle_key;size:7;uniqueIndex:uk_admin_host_agency_salary_policy_name;uniqueIndex:uk_admin_host_agency_salary_policy_version;index:idx_admin_host_agency_salary_policy_cycle,not null" json:"cycleKey"` PolicyVersion uint64 `gorm:"column:policy_version;uniqueIndex:uk_admin_host_agency_salary_policy_version;not null;default:0" json:"policyVersion"` Status string `gorm:"size:24;index:idx_admin_host_agency_salary_policy_region,not null;default:disabled" json:"status"` diff --git a/server/admin/internal/modules/hostagencypolicy/service.go b/server/admin/internal/modules/hostagencypolicy/service.go index bbc07dc4..42acbbdf 100644 --- a/server/admin/internal/modules/hostagencypolicy/service.go +++ b/server/admin/internal/modules/hostagencypolicy/service.go @@ -201,14 +201,12 @@ func (s *Service) Update(ctx context.Context, appCode string, actorID uint, id u if err != nil { return policyDTO{}, err } + originalPolicyVersion := item.PolicyVersion runtimePublished, err := s.runtimePolicySnapshotExists(ctx, item) if err != nil { return policyDTO{}, err } - if item.PublishStatus == publishStatusPublished || item.PublishedAtMS > 0 || runtimePublished { - // 运行侧已经按周期绑定该版本；编辑会让历史结算口径漂移，只能为新周期新建政策。 - return policyDTO{}, errors.New("已发布的周期政策版本不可修改，请为新周期创建政策") - } + publishedRevision := item.PublishStatus == publishStatusPublished || item.PublishedAtMS > 0 || runtimePublished if strings.TrimSpace(req.PolicyType) == "" { // Update DTO 的 policy_type 可选；旧前端没有该字段时必须保留草稿原类型，不能按创建默认值覆盖为工资型。 req.PolicyType = policyTypeForUpdate(req.PolicyType, item.PolicyType) @@ -218,6 +216,13 @@ func (s *Service) Update(ctx context.Context, appCode string, actorID uint, id u if err != nil { return policyDTO{}, err } + if publishedRevision { + // 已结束月份仍完全锁定；当前月和未来月只允许修改业务参数，发布身份不变，随后生成新的不可变运行版本。 + if err := validatePublishedPolicyRevision(item, updated, time.Now().UTC()); err != nil { + return policyDTO{}, err + } + item.PolicyVersion = nextHostSalaryPolicyVersion(item.PolicyVersion, time.Now().UTC()) + } item.Name = updated.Name item.PolicyType = updated.PolicyType item.RegionID = updated.RegionID @@ -240,7 +245,7 @@ func (s *Service) Update(ctx context.Context, appCode string, actorID uint, id u item.PlatformWithdrawalAllowedDays = updated.PlatformWithdrawalAllowedDays item.Description = updated.Description item.UpdatedByAdminID = actorID - // 编辑后运行快照可能已经落后于 admin 配置，因此保留上次发布时间但把状态重新标成 draft。 + // 保存修订只更新 admin 草稿；运营重新发布后 wallet 才原子写入新快照并切换当前周期绑定。 item.PublishStatus = publishStatusDraft item.PublishError = "" // 等级明细按整包替换处理：前端提交的是一套完整政策梯度，不做局部 patch，减少排序和递增口径漂移。 @@ -248,12 +253,49 @@ func (s *Service) Update(ctx context.Context, appCode string, actorID uint, id u if err := s.ensureNoActiveOverlap(item, id); err != nil { return policyDTO{}, err } - if err := s.store.UpdateHostAgencySalaryPolicy(&item); err != nil { + if err := s.store.UpdateHostAgencySalaryPolicy(&item, originalPolicyVersion); err != nil { return policyDTO{}, normalizePolicyPersistenceError(err) } return policyFromModel(item), nil } +func validatePublishedPolicyRevision(existing model.HostAgencySalaryPolicy, updated model.HostAgencySalaryPolicy, now time.Time) error { + currentCycle := now.UTC().Format("2006-01") + if existing.CycleKey < currentCycle { + return errors.New("已结束工资月的政策不可修改") + } + // 类型、月份和区域共同决定钱包记账分流及周期绑定；修订这些身份字段会留下无法解释的旧绑定，必须新建政策。 + if normalizePolicyType(existing.PolicyType) != normalizePolicyType(updated.PolicyType) || + existing.CycleKey != updated.CycleKey || !samePolicyRegionIDs(existing.RegionIDs, updated.RegionIDs) { + return errors.New("已发布政策的类型、周期和适用区域不可修改，请新建政策") + } + if updated.Status != policyStatusActive { + return errors.New("已发布政策修订时必须保持启用") + } + return nil +} + +func samePolicyRegionIDs(left []int64, right []int64) bool { + if len(left) != len(right) { + return false + } + for index := range left { + if left[index] != right[index] { + return false + } + } + return true +} + +func nextHostSalaryPolicyVersion(current uint64, now time.Time) uint64 { + // epoch ms 对 JSON 安全且按时间可读；同一行在仓储事务中用旧版本 CAS，毫秒内并发也只允许一个提交成功。 + candidate := uint64(now.UTC().UnixMilli()) + if candidate <= current { + return current + 1 + } + return candidate +} + func policyTypeForUpdate(requested string, existing string) string { if strings.TrimSpace(requested) == "" { return existing @@ -284,7 +326,7 @@ func (s *Service) Publish(ctx context.Context, appCode string, actorID uint, id if err != nil { return policyDTO{}, err } - if item.PublishStatus == publishStatusPublished || item.PublishedAtMS > 0 { + if item.PublishStatus == publishStatusPublished { // 重复点击发布按幂等成功返回；不再覆盖运行表，也不生成新的周期绑定。 return policyFromModel(item), nil } @@ -314,6 +356,9 @@ func normalizePolicyPersistenceError(err error) error { if errors.Is(err, repository.ErrHostAgencySalaryPolicyActiveScopeConflict) { return errors.New("同一区域同一工资周期只能存在一个启用政策") } + if errors.Is(err, repository.ErrHostAgencySalaryPolicyConcurrentUpdate) { + return errors.New("政策已被其他人修改，请刷新后重试") + } return err } @@ -382,13 +427,13 @@ func (s *Service) publishRuntimePolicy(ctx context.Context, item model.HostAgenc // 仅兼容迁移前创建但尚未发布的草稿；新记录在 admin 仓储创建事务中固定为 policy ID。 policyVersion = uint64(item.ID) } - // 跨库状态回写失败后重试时，wallet 可能已经拥有完整快照；内容一致视为幂等成功， - // 但任何业务字段或等级不一致都拒绝覆盖，继续保持已发布版本不可变。 + // 跨库状态回写失败后重试时，wallet 可能已经拥有同版本快照；内容一致视为幂等成功， + // 新修订使用新的 policy_version，因此旧版本始终保留且任何同版本内容都不能被覆盖。 if err := ensureRuntimeHostSalaryPolicySnapshot(ctx, tx, item, policyVersion, publishedAtMS); err != nil { return err } - // 每个区域各自拥有不可覆盖的周期绑定，但同一政策快照和等级只写入一次，避免多区域版本漂移。 - // 系统继承绑定的 source_cycle_key 指向上月，允许随后自动生成或人工创建的本月独立版本接管；显式绑定永不覆盖。 + // 每个区域各自拥有周期绑定；继承版本可由本月显式版本接管，同一政策的更高修订版本也可切换当前绑定。 + // 旧运行快照不删除，工资记录中的 policy_version 仍能准确指回当时使用的口径。 for _, regionID := range item.RegionIDs { if err := bindExplicitHostSalaryPolicy(ctx, tx, item, regionID, policyVersion, publishedAtMS); err != nil { return err @@ -431,7 +476,7 @@ type runtimeHostSalaryLevelSnapshot struct { } func ensureRuntimeHostSalaryPolicySnapshot(ctx context.Context, tx *sql.Tx, item model.HostAgencySalaryPolicy, policyVersion uint64, publishedAtMS int64) error { - // 先以完整快照尝试原子占位；主键已存在时只做 policy_id 自赋值，绝不改写任何业务字段。 + // 先以完整快照尝试原子占位；复合主键包含 policy_version，已存在时只做自赋值，绝不改写业务字段。 // 两个首次发布请求会在同一主键写锁上排队，避免“都 SELECT 到不存在，再 INSERT”形成 gap-lock 死锁。 if _, err := tx.ExecContext(ctx, ` INSERT INTO host_agency_salary_policies ( @@ -463,8 +508,8 @@ func ensureRuntimeHostSalaryPolicySnapshot(ctx context.Context, tx *sql.Tx, item coin_seller_withdrawal_limit_period, coin_seller_withdrawal_limit_count, platform_withdrawal_limit_period, platform_withdrawal_limit_count, platform_withdrawal_allowed_days FROM host_agency_salary_policies - WHERE app_code = ? AND policy_id = ? - FOR UPDATE`, item.AppCode, item.ID).Scan( + WHERE app_code = ? AND policy_id = ? AND policy_version = ? + FOR UPDATE`, item.AppCode, item.ID, policyVersion).Scan( &existing.CycleKey, &existing.PolicyVersion, &existing.Name, &existing.PolicyType, &existing.RegionID, &existing.Status, &existing.SettlementMode, &existing.SettlementTriggerMode, &existing.GiftCoinToDiamondRatio, &existing.PointDiamondsPerUSD, &existing.CoinsPerUSD, @@ -484,7 +529,7 @@ func ensureRuntimeHostSalaryPolicySnapshot(ctx context.Context, tx *sql.Tx, item if err != nil { return err } - existingLevels, err := loadRuntimeHostSalaryLevels(ctx, tx, item.AppCode, item.ID) + existingLevels, err := loadRuntimeHostSalaryLevels(ctx, tx, item.AppCode, item.ID, policyVersion) if err != nil { return err } @@ -495,17 +540,17 @@ func ensureRuntimeHostSalaryPolicySnapshot(ctx context.Context, tx *sql.Tx, item // 运行表保存美分整数，避免 wallet-service 结算时再次解析 decimal 字符串造成口径漂移。 if _, err := tx.ExecContext(ctx, ` INSERT INTO host_agency_salary_policy_levels ( - app_code, policy_id, level_no, required_diamonds, host_salary_usd_minor, + app_code, policy_id, policy_version, level_no, required_diamonds, host_salary_usd_minor, host_coin_reward, agency_salary_usd_minor, status, sort_order, created_at_ms, updated_at_ms - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, - item.AppCode, item.ID, level.LevelNo, level.RequiredDiamonds, level.HostSalaryUSDMinor, + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + item.AppCode, item.ID, policyVersion, level.LevelNo, level.RequiredDiamonds, level.HostSalaryUSDMinor, level.HostCoinReward, level.AgencySalaryUSDMinor, level.Status, level.SortOrder, publishedAtMS, publishedAtMS, ); err != nil { return err } } - existingLevels, err = loadRuntimeHostSalaryLevels(ctx, tx, item.AppCode, item.ID) + existingLevels, err = loadRuntimeHostSalaryLevels(ctx, tx, item.AppCode, item.ID, policyVersion) if err != nil { return err } @@ -554,13 +599,13 @@ func runtimeHostSalaryLevelsFromModel(levels []model.HostAgencySalaryLevel) ([]r return out, nil } -func loadRuntimeHostSalaryLevels(ctx context.Context, tx *sql.Tx, appCode string, policyID uint) ([]runtimeHostSalaryLevelSnapshot, error) { +func loadRuntimeHostSalaryLevels(ctx context.Context, tx *sql.Tx, appCode string, policyID uint, policyVersion uint64) ([]runtimeHostSalaryLevelSnapshot, error) { rows, err := tx.QueryContext(ctx, ` SELECT level_no, required_diamonds, host_salary_usd_minor, host_coin_reward, agency_salary_usd_minor, status, sort_order FROM host_agency_salary_policy_levels - WHERE app_code = ? AND policy_id = ? - ORDER BY level_no ASC`, appCode, policyID) + WHERE app_code = ? AND policy_id = ? AND policy_version = ? + ORDER BY level_no ASC`, appCode, policyID, policyVersion) if err != nil { return nil, err } @@ -613,6 +658,26 @@ func bindExplicitHostSalaryPolicy(ctx context.Context, tx *sql.Tx, item model.Ho if existingPolicyID == uint64(item.ID) && existingPolicyVersion == policyVersion { return nil } + if existingPolicyID == uint64(item.ID) && existingPolicyVersion < policyVersion && strings.TrimSpace(sourceCycleKey) == item.CycleKey { + // 当前/未来月份的同一逻辑政策允许发布更高版本；WHERE 带旧版本，防止并发发布把更新顺序倒置。 + result, updateErr := tx.ExecContext(ctx, ` + UPDATE host_salary_policy_cycle_bindings + SET policy_version = ?, created_at_ms = ? + WHERE app_code = ? AND region_id = ? AND cycle_key = ? AND policy_id = ? AND policy_version = ? AND source_cycle_key = cycle_key`, + policyVersion, publishedAtMS, item.AppCode, regionID, item.CycleKey, item.ID, existingPolicyVersion, + ) + if updateErr != nil { + return updateErr + } + affected, updateErr := result.RowsAffected() + if updateErr != nil { + return updateErr + } + if affected != 1 { + return errors.New("工资月政策版本已被并发修改，请重试") + } + return nil + } if strings.TrimSpace(sourceCycleKey) == "" || strings.TrimSpace(sourceCycleKey) == item.CycleKey { return errors.New("该区域工资月已经绑定其他显式政策版本") } diff --git a/server/admin/internal/modules/hostagencypolicy/service_test.go b/server/admin/internal/modules/hostagencypolicy/service_test.go index 1b0654da..acae71f8 100644 --- a/server/admin/internal/modules/hostagencypolicy/service_test.go +++ b/server/admin/internal/modules/hostagencypolicy/service_test.go @@ -3,6 +3,7 @@ package hostagencypolicy import ( "context" "testing" + "time" "hyapp-admin-server/internal/model" "hyapp-admin-server/internal/repository" @@ -242,12 +243,12 @@ func TestEnsureRuntimeHostSalaryPolicySnapshotAcceptsIdenticalExistingSnapshot(t Levels: []model.HostAgencySalaryLevel{{LevelNo: 1, RequiredDiamonds: 100, HostSalaryUSD: "1.50", HostCoinReward: 20, AgencySalaryUSD: "0.50", Status: policyStatusActive, SortOrder: 1}}, } mock.ExpectExec("INSERT INTO host_agency_salary_policies").WillReturnResult(sqlmock.NewResult(0, 0)) - mock.ExpectQuery("SELECT cycle_key, policy_version").WithArgs("fami", uint(81)).WillReturnRows(sqlmock.NewRows([]string{ + mock.ExpectQuery("SELECT cycle_key, policy_version").WithArgs("fami", uint(81), uint64(81)).WillReturnRows(sqlmock.NewRows([]string{ "cycle_key", "policy_version", "name", "policy_type", "region_id", "status", "settlement_mode", "settlement_trigger_mode", "gift_ratio", "points_per_usd", "coins_per_usd", "minimum_usd_minor", "fee_bps", "agency_bps", "residual_rate", "coin_period", "coin_count", "platform_period", "platform_count", "platform_allowed_days", }).AddRow("2026-08", 81, "August policy", "SALARY_DIAMOND", 25, "active", "daily", "automatic", "1.000000", 0, 0, 0, 0, 0, "0.000000000000", "week", 2, "day", 1, "15,30")) - mock.ExpectQuery("SELECT level_no, required_diamonds").WithArgs("fami", uint(81)).WillReturnRows(sqlmock.NewRows([]string{ + mock.ExpectQuery("SELECT level_no, required_diamonds").WithArgs("fami", uint(81), uint64(81)).WillReturnRows(sqlmock.NewRows([]string{ "level_no", "required_diamonds", "host_salary_usd_minor", "host_coin_reward", "agency_salary_usd_minor", "status", "sort_order", }).AddRow(1, 100, 150, 20, 50, "active", 1)) if err := ensureRuntimeHostSalaryPolicySnapshot(context.Background(), tx, item, 81, 200); err != nil { @@ -281,6 +282,34 @@ func TestRuntimeHostSalaryPolicySnapshotRejectsDifferentContent(t *testing.T) { } } +func TestBindExplicitHostSalaryPolicyMovesSamePolicyToHigherVersion(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("create sqlmock: %v", err) + } + defer db.Close() + mock.ExpectBegin() + tx, err := db.BeginTx(context.Background(), nil) + if err != nil { + t.Fatalf("begin transaction: %v", err) + } + item := model.HostAgencySalaryPolicy{ID: 81, AppCode: "fami", CycleKey: "2026-07"} + mock.ExpectQuery("SELECT policy_id, policy_version, source_cycle_key"). + WithArgs("fami", int64(25), "2026-07"). + WillReturnRows(sqlmock.NewRows([]string{"policy_id", "policy_version", "source_cycle_key"}).AddRow(81, 81, "2026-07")) + mock.ExpectExec("UPDATE host_salary_policy_cycle_bindings"). + WithArgs(uint64(1784650000000), int64(1784650001000), "fami", int64(25), "2026-07", uint(81), uint64(81)). + WillReturnResult(sqlmock.NewResult(0, 1)) + if err := bindExplicitHostSalaryPolicy(context.Background(), tx, item, 25, 1784650000000, 1784650001000); err != nil { + t.Fatalf("same current policy should move to its higher immutable version: %v", err) + } + mock.ExpectRollback() + _ = tx.Rollback() + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("sql expectations: %v", err) + } +} + func TestPublishFailureUpdateUsesNotPublishedCAS(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { @@ -292,7 +321,7 @@ func TestPublishFailureUpdateUsesNotPublishedCAS(t *testing.T) { t.Fatalf("open gorm: %v", err) } store := repository.New(gormDB) - mock.ExpectExec("UPDATE .*admin_host_agency_salary_policies.*publish_status <> \\? AND published_at_ms = 0"). + mock.ExpectExec("UPDATE .*admin_host_agency_salary_policies.*publish_status <> \\?"). WillReturnResult(sqlmock.NewResult(0, 0)) if err := store.UpdateHostAgencySalaryPolicyPublishFailureIfNotPublished("fami", 91, "duplicate publish", 7); err != nil { t.Fatalf("conditional failure update: %v", err) @@ -310,3 +339,33 @@ func TestPolicyTypeForUpdatePreservesPointDraftWhenFieldIsOmitted(t *testing.T) t.Fatalf("explicit draft type change must remain visible to validation, got %q", got) } } + +func TestPublishedPolicyRevisionAllowsCurrentBusinessParametersButLocksIdentity(t *testing.T) { + existing := model.HostAgencySalaryPolicy{ + PolicyType: policyTypePointDiamond, CycleKey: "2026-07", RegionID: 25, RegionIDs: []int64{25, 26}, Status: policyStatusActive, + AgencyPointShareBPS: 20, + } + updated := existing + updated.AgencyPointShareBPS = 2000 + if err := validatePublishedPolicyRevision(existing, updated, time.Date(2026, time.July, 22, 0, 0, 0, 0, time.UTC)); err != nil { + t.Fatalf("current policy business parameters should be revisable: %v", err) + } + updated.RegionIDs = []int64{25} + if err := validatePublishedPolicyRevision(existing, updated, time.Date(2026, time.July, 22, 0, 0, 0, 0, time.UTC)); err == nil { + t.Fatal("published policy region identity must stay locked") + } + updated = existing + if err := validatePublishedPolicyRevision(existing, updated, time.Date(2026, time.August, 1, 0, 0, 0, 0, time.UTC)); err == nil { + t.Fatal("closed-cycle policy must remain immutable") + } +} + +func TestNextHostSalaryPolicyVersionIsMonotonicAndJSONSafe(t *testing.T) { + now := time.Date(2026, time.July, 22, 0, 0, 0, 0, time.UTC) + if got := nextHostSalaryPolicyVersion(81, now); got != uint64(now.UnixMilli()) { + t.Fatalf("revision should use readable epoch-ms version, got %d", got) + } + if got := nextHostSalaryPolicyVersion(uint64(now.UnixMilli()), now); got != uint64(now.UnixMilli())+1 { + t.Fatalf("same-millisecond revision must still increase, got %d", got) + } +} diff --git a/server/admin/internal/modules/hostsalarysettlement/service.go b/server/admin/internal/modules/hostsalarysettlement/service.go index 308a912d..6d1edcd7 100644 --- a/server/admin/internal/modules/hostsalarysettlement/service.go +++ b/server/admin/internal/modules/hostsalarysettlement/service.go @@ -556,7 +556,7 @@ func (s *Service) resolvePolicy(ctx context.Context, appCode string, regionID in if strings.TrimSpace(policy.PolicyType) != policyTypeSalaryDiamond { return hostPolicy{}, false, fmt.Errorf("policy type %s is not eligible for salary settlement", policy.PolicyType) } - levels, err := s.policyLevels(ctx, appCode, policy.PolicyID) + levels, err := s.policyLevels(ctx, appCode, policy.PolicyID, policy.PolicyVersion) if err != nil { return hostPolicy{}, false, err } @@ -564,12 +564,12 @@ func (s *Service) resolvePolicy(ctx context.Context, appCode string, regionID in return policy, true, nil } -func (s *Service) policyLevels(ctx context.Context, appCode string, policyID uint64) ([]hostPolicyLevel, error) { +func (s *Service) policyLevels(ctx context.Context, appCode string, policyID uint64, policyVersion uint64) ([]hostPolicyLevel, error) { rows, err := s.walletDB.QueryContext(ctx, ` SELECT level_no, required_diamonds, host_salary_usd_minor, host_coin_reward, agency_salary_usd_minor FROM host_agency_salary_policy_levels - WHERE app_code = ? AND policy_id = ? AND status = 'active' - ORDER BY required_diamonds ASC, level_no ASC`, appCode, policyID) + WHERE app_code = ? AND policy_id = ? AND policy_version = ? AND status = 'active' + ORDER BY required_diamonds ASC, level_no ASC`, appCode, policyID, policyVersion) if err != nil { return nil, err } diff --git a/server/admin/internal/repository/host_agency_salary_policy_repository.go b/server/admin/internal/repository/host_agency_salary_policy_repository.go index e4faa0b0..605dc4a1 100644 --- a/server/admin/internal/repository/host_agency_salary_policy_repository.go +++ b/server/admin/internal/repository/host_agency_salary_policy_repository.go @@ -9,9 +9,13 @@ import ( mysqlDriver "github.com/go-sql-driver/mysql" "gorm.io/gorm" + "gorm.io/gorm/clause" ) -var ErrHostAgencySalaryPolicyActiveScopeConflict = errors.New("host agency salary policy active scope conflict") +var ( + ErrHostAgencySalaryPolicyActiveScopeConflict = errors.New("host agency salary policy active scope conflict") + ErrHostAgencySalaryPolicyConcurrentUpdate = errors.New("host agency salary policy was updated concurrently") +) type HostAgencySalaryPolicyListOptions struct { AppCode string @@ -167,8 +171,17 @@ func (s *Store) CreateHostAgencySalaryPolicy(item *model.HostAgencySalaryPolicy) }) } -func (s *Store) UpdateHostAgencySalaryPolicy(item *model.HostAgencySalaryPolicy) error { +func (s *Store) UpdateHostAgencySalaryPolicy(item *model.HostAgencySalaryPolicy, expectedPolicyVersion uint64) error { return s.db.Transaction(func(tx *gorm.DB) error { + // 同一已发布政策可在当前月生成新版本；先锁主行并校验调用方读到的版本，避免两个运营提交互相覆盖配置和等级。 + var locked model.HostAgencySalaryPolicy + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Select("id", "policy_version"). + Where("app_code = ? AND id = ?", item.AppCode, item.ID).First(&locked).Error; err != nil { + return err + } + if locked.PolicyVersion != expectedPolicyVersion { + return ErrHostAgencySalaryPolicyConcurrentUpdate + } // 等级是整包配置，不做逐条 diff；主表保存成功后删除旧明细，再插入新梯度。 levels := append([]model.HostAgencySalaryLevel(nil), item.Levels...) item.Levels = nil @@ -214,9 +227,9 @@ func (s *Store) UpdateHostAgencySalaryPolicyPublishState(appCode string, id uint func (s *Store) UpdateHostAgencySalaryPolicyPublishFailureIfNotPublished(appCode string, id uint, publishError string, actorID uint) error { // wallet 提交与 admin 状态回写跨库，失败回写必须带状态条件：并发成功发布已经写入 published 后， - // 较晚返回的重复发布错误不能把真实成功状态覆盖成 failed，也不能清空成功时间。 + // 较晚返回的重复发布错误不能把真实成功状态覆盖成 failed；published_at_ms 保留上次成功时间，表明该政策身份已经发布过。 return s.db.Model(&model.HostAgencySalaryPolicy{}). - Where("app_code = ? AND id = ? AND publish_status <> ? AND published_at_ms = 0", strings.TrimSpace(appCode), id, "published"). + Where("app_code = ? AND id = ? AND publish_status <> ?", strings.TrimSpace(appCode), id, "published"). Updates(map[string]any{ "publish_status": "failed", "publish_error": strings.TrimSpace(publishError), diff --git a/services/gateway-service/internal/transport/http/activityapi/handler.go b/services/gateway-service/internal/transport/http/activityapi/handler.go index 1183e589..3024311a 100644 --- a/services/gateway-service/internal/transport/http/activityapi/handler.go +++ b/services/gateway-service/internal/transport/http/activityapi/handler.go @@ -4,6 +4,7 @@ import ( "context" userv1 "hyapp.local/api/proto/user/v1" + walletv1 "hyapp.local/api/proto/wallet/v1" "hyapp/pkg/userleaderboard" "hyapp/services/gateway-service/internal/appconfig" "hyapp/services/gateway-service/internal/client" @@ -18,7 +19,7 @@ type Handler struct { achievementClient client.AchievementClient userProfileClient UserProfileReader roomQueryClient client.RoomQueryClient - walletClient client.WalletClient + walletClient ActivityWalletClient userLeaderboard UserLeaderboardStore runtimeCapabilities appconfig.RuntimeCapabilityReader registrationReward client.RegistrationRewardClient @@ -49,6 +50,15 @@ type UserProfileReader interface { BatchGetUsers(ctx context.Context, req *userv1.BatchGetUsersRequest) (*userv1.BatchGetUsersResponse, error) } +// ActivityWalletClient 只开放活动模块现有的礼物目录、转盘扣款和榜单展示能力，避免 HTTP handler 获得完整账务接口。 +type ActivityWalletClient interface { + BatchGetUserEquippedResources(ctx context.Context, req *walletv1.BatchGetUserEquippedResourcesRequest) (*walletv1.BatchGetUserEquippedResourcesResponse, error) + GetResource(ctx context.Context, req *walletv1.GetResourceRequest) (*walletv1.GetResourceResponse, error) + GetResourceGroup(ctx context.Context, req *walletv1.GetResourceGroupRequest) (*walletv1.GetResourceGroupResponse, error) + ListGiftConfigs(ctx context.Context, req *walletv1.ListGiftConfigsRequest) (*walletv1.ListGiftConfigsResponse, error) + DebitWheelDraw(ctx context.Context, req *walletv1.DebitWheelDrawRequest) (*walletv1.DebitWheelDrawResponse, error) +} + type Config struct { TaskClient client.TaskClient GrowthLevelClient client.GrowthLevelClient diff --git a/services/gateway-service/internal/transport/http/activityapi/user_leaderboard_handler.go b/services/gateway-service/internal/transport/http/activityapi/user_leaderboard_handler.go index 4126d010..fe9119dc 100644 --- a/services/gateway-service/internal/transport/http/activityapi/user_leaderboard_handler.go +++ b/services/gateway-service/internal/transport/http/activityapi/user_leaderboard_handler.go @@ -2,6 +2,7 @@ package activityapi import ( "encoding/json" + "errors" "net/http" "strconv" "strings" @@ -48,6 +49,7 @@ type userLeaderboardResponse struct { type userLeaderboardItem struct { Rank int64 `json:"rank"` UserID string `json:"user_id,omitempty"` + IdentityHidden bool `json:"identity_hidden,omitempty"` RoomID string `json:"room_id,omitempty"` GiftValue int64 `json:"gift_value"` GiftCount int64 `json:"gift_count"` @@ -284,17 +286,19 @@ func (h *Handler) enrichUserLeaderboardItems(request *http.Request, boardType st if err != nil { return err } - shortBadgeCovers, err := h.userLeaderboardShortBadgeCovers(request, userIDs) + presentation, err := h.userLeaderboardPresentation(request, userIDs) if err != nil { return err } for i := range items { applyUserLeaderboardProfile(&items[i], profiles) - applyUserLeaderboardShortBadgeCovers(&items[i], shortBadgeCovers) + applyUserLeaderboardShortBadgeCovers(&items[i], presentation.ShortBadgeCovers) + applyUserLeaderboardPrivacy(&items[i], presentation.LeaderboardInvisible) } if myRank != nil { applyUserLeaderboardProfile(myRank, profiles) - applyUserLeaderboardShortBadgeCovers(myRank, shortBadgeCovers) + applyUserLeaderboardShortBadgeCovers(myRank, presentation.ShortBadgeCovers) + applyUserLeaderboardPrivacy(myRank, presentation.LeaderboardInvisible) } return nil } @@ -392,28 +396,47 @@ func applyUserLeaderboardProfile(item *userLeaderboardItem, profiles map[int64]* } } -func (h *Handler) userLeaderboardShortBadgeCovers(request *http.Request, userIDs []int64) (map[int64][]string, error) { - result := make(map[int64][]string) +type userLeaderboardPresentation struct { + ShortBadgeCovers map[int64][]string + LeaderboardInvisible map[int64]struct{} +} + +func (h *Handler) userLeaderboardPresentation(request *http.Request, userIDs []int64) (userLeaderboardPresentation, error) { + result := userLeaderboardPresentation{ + ShortBadgeCovers: make(map[int64][]string), + LeaderboardInvisible: make(map[int64]struct{}), + } userIDs = uniqueUserLeaderboardInt64s(userIDs) - if h.walletClient == nil || len(userIDs) == 0 { - // 榜单基础资料不依赖装扮服务；未接 wallet 的环境保持兼容，只是不返回短徽章。 + if len(userIDs) == 0 { return result, nil } - // 一页榜单只发起一次批量查询，避免按用户逐个读取权益和资源造成 N+1 放大。 + if h.walletClient == nil { + // 隐私权益不能在 wallet 未接入时 fail-open，否则 VIP 用户会在依赖故障期间暴露身份。 + return userLeaderboardPresentation{}, errors.New("wallet client is required for leaderboard privacy") + } + // 一页榜单只发起一次批量查询：静态短徽章与最终 VIP 隐私权限使用同一批用户，避免 N+1 放大。 resp, err := h.walletClient.BatchGetUserEquippedResources(request.Context(), &walletv1.BatchGetUserEquippedResourcesRequest{ - RequestId: httpkit.RequestIDFromContext(request.Context()), - AppCode: appcode.FromContext(request.Context()), - UserIds: userIDs, - ResourceTypes: []string{"badge"}, + RequestId: httpkit.RequestIDFromContext(request.Context()), + AppCode: appcode.FromContext(request.Context()), + UserIds: userIDs, + ResourceTypes: []string{"badge"}, + VipBenefitCodes: []string{"leaderboard_invisible"}, }) if err != nil { - return nil, err + return userLeaderboardPresentation{}, err } - return userLeaderboardShortBadgeCoversFromResponse(resp), nil + return userLeaderboardPresentationFromResponse(resp), nil } func userLeaderboardShortBadgeCoversFromResponse(resp *walletv1.BatchGetUserEquippedResourcesResponse) map[int64][]string { - result := make(map[int64][]string) + return userLeaderboardPresentationFromResponse(resp).ShortBadgeCovers +} + +func userLeaderboardPresentationFromResponse(resp *walletv1.BatchGetUserEquippedResourcesResponse) userLeaderboardPresentation { + result := userLeaderboardPresentation{ + ShortBadgeCovers: make(map[int64][]string), + LeaderboardInvisible: make(map[int64]struct{}), + } if resp == nil { return result } @@ -421,6 +444,12 @@ func userLeaderboardShortBadgeCoversFromResponse(resp *walletv1.BatchGetUserEqui if user.GetUserId() <= 0 { continue } + for _, code := range user.GetAllowedVipBenefitCodes() { + if strings.EqualFold(strings.TrimSpace(code), "leaderboard_invisible") { + result.LeaderboardInvisible[user.GetUserId()] = struct{}{} + break + } + } seen := make(map[string]struct{}) for _, entitlement := range user.GetResources() { resource := entitlement.GetResource() @@ -436,12 +465,29 @@ func userLeaderboardShortBadgeCoversFromResponse(resp *walletv1.BatchGetUserEqui continue } seen[coverURL] = struct{}{} - result[user.GetUserId()] = append(result[user.GetUserId()], coverURL) + result.ShortBadgeCovers[user.GetUserId()] = append(result.ShortBadgeCovers[user.GetUserId()], coverURL) } } return result } +func applyUserLeaderboardPrivacy(item *userLeaderboardItem, invisible map[int64]struct{}) { + if item == nil { + return + } + userID, err := strconv.ParseInt(item.UserID, 10, 64) + if err != nil || userID <= 0 { + return + } + if _, hidden := invisible[userID]; !hidden { + return + } + // 名次和分值仍是榜单事实；身份字段必须整体清空，避免 H5 禁止跳转后响应里仍泄露内部 ID 或资料。 + item.UserID = "" + item.User = nil + item.IdentityHidden = true +} + func isUserLeaderboardShortBadge(rawMetadata string) bool { metadata := map[string]any{} if err := json.Unmarshal([]byte(strings.TrimSpace(rawMetadata)), &metadata); err != nil { diff --git a/services/gateway-service/internal/transport/http/activityapi/user_leaderboard_handler_test.go b/services/gateway-service/internal/transport/http/activityapi/user_leaderboard_handler_test.go index 22607c58..736773cb 100644 --- a/services/gateway-service/internal/transport/http/activityapi/user_leaderboard_handler_test.go +++ b/services/gateway-service/internal/transport/http/activityapi/user_leaderboard_handler_test.go @@ -18,6 +18,35 @@ import ( walletv1 "hyapp.local/api/proto/wallet/v1" ) +type fakeUserLeaderboardPresentationReader struct { + last *walletv1.BatchGetUserEquippedResourcesRequest + response *walletv1.BatchGetUserEquippedResourcesResponse +} + +func (*fakeUserLeaderboardPresentationReader) GetResource(context.Context, *walletv1.GetResourceRequest) (*walletv1.GetResourceResponse, error) { + return &walletv1.GetResourceResponse{}, nil +} + +func (*fakeUserLeaderboardPresentationReader) GetResourceGroup(context.Context, *walletv1.GetResourceGroupRequest) (*walletv1.GetResourceGroupResponse, error) { + return &walletv1.GetResourceGroupResponse{}, nil +} + +func (*fakeUserLeaderboardPresentationReader) ListGiftConfigs(context.Context, *walletv1.ListGiftConfigsRequest) (*walletv1.ListGiftConfigsResponse, error) { + return &walletv1.ListGiftConfigsResponse{}, nil +} + +func (*fakeUserLeaderboardPresentationReader) DebitWheelDraw(context.Context, *walletv1.DebitWheelDrawRequest) (*walletv1.DebitWheelDrawResponse, error) { + return &walletv1.DebitWheelDrawResponse{}, nil +} + +func (f *fakeUserLeaderboardPresentationReader) BatchGetUserEquippedResources(_ context.Context, request *walletv1.BatchGetUserEquippedResourcesRequest) (*walletv1.BatchGetUserEquippedResourcesResponse, error) { + f.last = request + if f.response != nil { + return f.response, nil + } + return &walletv1.BatchGetUserEquippedResourcesResponse{}, nil +} + func TestUserLeaderboardPeriodWindowUsesUTCBoundaries(t *testing.T) { now := time.Date(2026, 5, 20, 15, 30, 0, 0, time.FixedZone("CST", 8*60*60)) @@ -57,7 +86,8 @@ func TestNormalizeUserLeaderboardTypeAndPeriod(t *testing.T) { func TestUserLeaderboardShortBadgeCoversFromResponse(t *testing.T) { response := &walletv1.BatchGetUserEquippedResourcesResponse{Users: []*walletv1.UserEquippedResources{{ - UserId: 20002, + UserId: 20002, + AllowedVipBenefitCodes: []string{"leaderboard_invisible"}, Resources: []*walletv1.UserResourceEntitlement{ {Resource: &walletv1.Resource{ResourceId: 1, ResourceType: "badge", PreviewUrl: "https://cdn.example/short.png", MetadataJson: `{"badge_form":"tile","badge_kind":"normal"}`}}, {Resource: &walletv1.Resource{ResourceId: 2, ResourceType: "badge", PreviewUrl: "https://cdn.example/long.png", MetadataJson: `{"badge_form":"strip","badge_kind":"normal"}`}}, @@ -74,6 +104,15 @@ func TestUserLeaderboardShortBadgeCoversFromResponse(t *testing.T) { if len(covers[99999]) != 0 { t.Fatalf("users without a short badge must stay absent: %+v", covers) } + presentation := userLeaderboardPresentationFromResponse(response) + if _, invisible := presentation.LeaderboardInvisible[20002]; !invisible { + t.Fatalf("wallet's final leaderboard privacy grant must be preserved: %+v", presentation) + } + item := userLeaderboardItem{UserID: "20002", User: &userLeaderboardUser{UserID: "20002", DisplayUserID: "251145", Username: "VIP"}} + applyUserLeaderboardPrivacy(&item, presentation.LeaderboardInvisible) + if !item.IdentityHidden || item.UserID != "" || item.User != nil { + t.Fatalf("invisible leaderboard identity must be removed without changing rank facts: %+v", item) + } } func TestRoomLeaderboardRoomFromSnapshotIncludesAvatar(t *testing.T) { @@ -147,9 +186,14 @@ func TestListUserLeaderboardsReadsRedisStore(t *testing.T) { ActivityLeaderboardScope: appconfig.ActivityLeaderboardScopeViewerRegion, ResourceEquipCommandIDPolicy: appconfig.ResourceEquipCommandIDPolicyLegacyOptional, }} + walletReader := &fakeUserLeaderboardPresentationReader{response: &walletv1.BatchGetUserEquippedResourcesResponse{Users: []*walletv1.UserEquippedResources{ + {UserId: 20002, AllowedVipBenefitCodes: []string{"leaderboard_invisible"}}, + {UserId: 10001}, + }}} handler := &Handler{ userLeaderboard: store, runtimeCapabilities: capabilityReader, + walletClient: walletReader, userProfileClient: &fakeInviteActivityUserProfileClient{user: &userv1.User{ AppCode: "hyapp_room", RegionId: 1001, }}, @@ -181,9 +225,12 @@ func TestListUserLeaderboardsReadsRedisStore(t *testing.T) { if envelope.Code != "OK" || envelope.Data.Total != 1 || len(envelope.Data.Items) != 1 { t.Fatalf("response mismatch: %+v", envelope) } - if envelope.Data.Items[0].UserID != "20002" || envelope.Data.Items[0].GiftValue != 900 || envelope.Data.MyRank == nil || envelope.Data.MyRank.Rank != 9 { + if !envelope.Data.Items[0].IdentityHidden || envelope.Data.Items[0].UserID != "" || envelope.Data.Items[0].User != nil || envelope.Data.Items[0].GiftValue != 900 || envelope.Data.MyRank == nil || envelope.Data.MyRank.Rank != 9 { t.Fatalf("leaderboard data mismatch: %+v my=%+v", envelope.Data.Items, envelope.Data.MyRank) } + if len(walletReader.last.GetVipBenefitCodes()) != 1 || walletReader.last.GetVipBenefitCodes()[0] != "leaderboard_invisible" { + t.Fatalf("leaderboard privacy must use the wallet batch gate: %+v", walletReader.last) + } } func TestListUserLeaderboardsAcceptsGameBoard(t *testing.T) { @@ -217,6 +264,7 @@ func TestListUserLeaderboardsAcceptsGameBoard(t *testing.T) { }} handler := &Handler{ userLeaderboard: store, + walletClient: &fakeUserLeaderboardPresentationReader{}, runtimeCapabilities: &fakeRuntimeCapabilityReader{capabilities: appconfig.RuntimeCapabilities{ ActivityLeaderboardScope: appconfig.ActivityLeaderboardScopeAppGlobal, ResourceEquipCommandIDPolicy: appconfig.ResourceEquipCommandIDPolicyLegacyOptional, diff --git a/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql b/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql index 0134e9fd..061b4b41 100644 --- a/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql +++ b/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql @@ -398,7 +398,7 @@ CREATE TABLE IF NOT EXISTS host_agency_salary_policies ( app_code VARCHAR(32) NOT NULL DEFAULT 'lalu' COMMENT '应用编码，用于多租户隔离', policy_id BIGINT UNSIGNED NOT NULL COMMENT '工资政策 ID，来自后台配置', cycle_key VARCHAR(7) NOT NULL COMMENT '政策归属工资周期，UTC yyyy-MM', - policy_version BIGINT UNSIGNED NOT NULL COMMENT '不可变政策版本，等于后台政策 ID', + policy_version BIGINT UNSIGNED NOT NULL COMMENT '不可变政策版本；首次等于后台政策 ID，修订后使用递增版本', name VARCHAR(120) NOT NULL DEFAULT '' COMMENT '政策名称', policy_type VARCHAR(32) NOT NULL DEFAULT 'SALARY_DIAMOND' COMMENT '政策类型：SALARY_DIAMOND/POINT_DIAMOND', region_id BIGINT NOT NULL COMMENT '适用区域 ID', @@ -421,7 +421,7 @@ CREATE TABLE IF NOT EXISTS host_agency_salary_policies ( effective_to_ms BIGINT NOT NULL DEFAULT 0 COMMENT '生效结束时间，0 表示长期有效', created_at_ms BIGINT NOT NULL COMMENT '创建时间，UTC epoch ms', updated_at_ms BIGINT NOT NULL COMMENT '更新时间，UTC epoch ms', - PRIMARY KEY (app_code, policy_id), + PRIMARY KEY (app_code, policy_id, policy_version), KEY idx_host_agency_salary_policy_cycle (app_code, region_id, cycle_key, status, settlement_mode, settlement_trigger_mode) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='主播代理工资政策运行快照表'; @@ -459,7 +459,7 @@ CREATE TABLE IF NOT EXISTS point_withdrawal_limit_reservations ( UNIQUE KEY uk_point_withdrawal_limit_reservation_command (app_code, source_asset_type, freeze_command_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='平台积分提现次数预留表'; --- 周期绑定一旦插入不可更新；复合主键保证新月份政策无法覆盖历史月份结算口径。 +-- 周期绑定按月唯一；历史月不更新，当前/未来月同一逻辑政策发布更高版本时只切换版本号。 CREATE TABLE IF NOT EXISTS host_salary_policy_cycle_bindings ( app_code VARCHAR(32) NOT NULL DEFAULT 'lalu' COMMENT '应用编码，用于多租户隔离', region_id BIGINT NOT NULL COMMENT '适用区域 ID', @@ -474,6 +474,7 @@ CREATE TABLE IF NOT EXISTS host_salary_policy_cycle_bindings ( CREATE TABLE IF NOT EXISTS host_agency_salary_policy_levels ( app_code VARCHAR(32) NOT NULL DEFAULT 'lalu' COMMENT '应用编码，用于多租户隔离', policy_id BIGINT UNSIGNED NOT NULL COMMENT '工资政策 ID', + policy_version BIGINT UNSIGNED NOT NULL COMMENT '所属不可变政策版本', level_no INT NOT NULL COMMENT '等级', required_diamonds BIGINT NOT NULL COMMENT '达到该等级所需主播当月钻石', host_salary_usd_minor BIGINT NOT NULL DEFAULT 0 COMMENT '主播累计美元工资，单位美分', @@ -483,8 +484,8 @@ CREATE TABLE IF NOT EXISTS host_agency_salary_policy_levels ( sort_order INT NOT NULL DEFAULT 0 COMMENT '排序权重', created_at_ms BIGINT NOT NULL COMMENT '创建时间，UTC epoch ms', updated_at_ms BIGINT NOT NULL COMMENT '更新时间，UTC epoch ms', - PRIMARY KEY (app_code, policy_id, level_no), - KEY idx_host_agency_salary_policy_level_required (app_code, policy_id, status, required_diamonds) + PRIMARY KEY (app_code, policy_id, policy_version, level_no), + KEY idx_host_agency_salary_policy_level_required (app_code, policy_id, policy_version, status, required_diamonds) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='主播代理工资政策等级运行快照表'; -- 结算进度按主播+月份独立记录累计已发值；日结/半月结只更新累计，月底清算只打 cleared 标记。 diff --git a/services/wallet-service/deploy/mysql/migrations/018_host_salary_policy_revisions.sql b/services/wallet-service/deploy/mysql/migrations/018_host_salary_policy_revisions.sql new file mode 100644 index 00000000..102bd01e --- /dev/null +++ b/services/wallet-service/deploy/mysql/migrations/018_host_salary_policy_revisions.sql @@ -0,0 +1,64 @@ +SET NAMES utf8mb4 COLLATE utf8mb4_unicode_ci; +SET time_zone = '+00:00'; + +USE hyapp_wallet; + +-- 两张政策配置表数据量远小于钱包流水；迁移只重建配置表主键并按既有政策点查回填等级版本， +-- 不扫描余额、交易、工资进度或结算记录。上线前仍应确认表行数，并在低峰执行主键 DDL。 +SET @ddl := IF( + (SELECT COUNT(*) FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'host_agency_salary_policy_levels' AND COLUMN_NAME = 'policy_version') = 0, + 'ALTER TABLE host_agency_salary_policy_levels ADD COLUMN policy_version BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT ''所属不可变政策版本'' AFTER policy_id, ALGORITHM=INPLACE, LOCK=NONE', + 'SELECT 1' +); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +-- 旧结构每个 policy_id 只有一个运行快照，按复合主键点关联即可回填，不读取任何业务流水。 +UPDATE host_agency_salary_policy_levels level +JOIN host_agency_salary_policies policy + ON policy.app_code = level.app_code + AND policy.policy_id = level.policy_id +SET level.policy_version = policy.policy_version +WHERE level.policy_version = 0; + +SET @ddl := IF( + COALESCE((SELECT GROUP_CONCAT(COLUMN_NAME ORDER BY SEQ_IN_INDEX SEPARATOR ',') FROM information_schema.STATISTICS + WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'host_agency_salary_policies' AND INDEX_NAME = 'PRIMARY'), '') <> 'app_code,policy_id,policy_version', + 'ALTER TABLE host_agency_salary_policies DROP PRIMARY KEY, ADD PRIMARY KEY (app_code, policy_id, policy_version), ALGORITHM=INPLACE, LOCK=NONE', + 'SELECT 1' +); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @ddl := IF( + COALESCE((SELECT GROUP_CONCAT(COLUMN_NAME ORDER BY SEQ_IN_INDEX SEPARATOR ',') FROM information_schema.STATISTICS + WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'host_agency_salary_policy_levels' AND INDEX_NAME = 'PRIMARY'), '') <> 'app_code,policy_id,policy_version,level_no', + 'ALTER TABLE host_agency_salary_policy_levels DROP PRIMARY KEY, ADD PRIMARY KEY (app_code, policy_id, policy_version, level_no), ALGORITHM=INPLACE, LOCK=NONE', + 'SELECT 1' +); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @level_required_index_columns := COALESCE(( + SELECT GROUP_CONCAT(COLUMN_NAME ORDER BY SEQ_IN_INDEX SEPARATOR ',') + FROM information_schema.STATISTICS + WHERE TABLE_SCHEMA = DATABASE() + AND TABLE_NAME = 'host_agency_salary_policy_levels' + AND INDEX_NAME = 'idx_host_agency_salary_policy_level_required' +), ''); +SET @ddl := IF( + @level_required_index_columns = 'app_code,policy_id,policy_version,status,required_diamonds', + 'SELECT 1', + IF( + @level_required_index_columns = '', + 'ALTER TABLE host_agency_salary_policy_levels ADD INDEX idx_host_agency_salary_policy_level_required (app_code, policy_id, policy_version, status, required_diamonds), ALGORITHM=INPLACE, LOCK=NONE', + 'ALTER TABLE host_agency_salary_policy_levels DROP INDEX idx_host_agency_salary_policy_level_required, ADD INDEX idx_host_agency_salary_policy_level_required (app_code, policy_id, policy_version, status, required_diamonds), ALGORITHM=INPLACE, LOCK=NONE' + ) +); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; diff --git a/services/wallet-service/internal/domain/resource/entitlement.go b/services/wallet-service/internal/domain/resource/entitlement.go index 5a576926..f25049e8 100644 --- a/services/wallet-service/internal/domain/resource/entitlement.go +++ b/services/wallet-service/internal/domain/resource/entitlement.go @@ -73,12 +73,14 @@ type UnequipUserResourceResult struct { } type BatchGetUserEquippedResourcesQuery struct { - AppCode string - UserIDs []int64 - ResourceTypes []string + AppCode string + UserIDs []int64 + ResourceTypes []string + VIPBenefitCodes []string } type UserEquippedResources struct { - UserID int64 - Resources []UserResourceEntitlement + UserID int64 + Resources []UserResourceEntitlement + AllowedVIPBenefitCodes []string } diff --git a/services/wallet-service/internal/service/wallet/vip_user_settings_integration_test.go b/services/wallet-service/internal/service/wallet/vip_user_settings_integration_test.go index 4a6b77c1..358af05f 100644 --- a/services/wallet-service/internal/service/wallet/vip_user_settings_integration_test.go +++ b/services/wallet-service/internal/service/wallet/vip_user_settings_integration_test.go @@ -7,6 +7,7 @@ import ( "hyapp/pkg/appcode" "hyapp/services/wallet-service/internal/domain/ledger" + resourcedomain "hyapp/services/wallet-service/internal/domain/resource" walletservice "hyapp/services/wallet-service/internal/service/wallet" "hyapp/services/wallet-service/internal/testutil/mysqltest" ) @@ -128,6 +129,22 @@ func TestVipUserSettingsDefaultPartialUpdateAndAppScope(t *testing.T) { if err != nil || !roomCheck.Allowed { t.Fatalf("enabled paid room entry benefit must be allowed: result=%+v err=%v", roomCheck, err) } + leaderboardInvisibleOn := true + if _, err := service.UpdateVipUserSettings(famiCtx, ledger.UpdateVipUserSettingsCommand{ + AppCode: "fami", UserID: paidUserID, LeaderboardInvisibleEnabled: &leaderboardInvisibleOn, + }); err != nil { + t.Fatalf("enable paid leaderboard privacy failed: %v", err) + } + presentation, err := service.BatchGetUserEquippedResources(famiCtx, resourcedomain.BatchGetUserEquippedResourcesQuery{ + AppCode: "fami", UserIDs: []int64{paidUserID, concurrentUserID}, + ResourceTypes: []string{resourcedomain.TypeBadge}, VIPBenefitCodes: []string{ledger.VipBenefitCodeLeaderboardInvisible}, + }) + if err != nil { + t.Fatalf("batch evaluate leaderboard privacy failed: %v", err) + } + if len(presentation) != 2 || len(presentation[0].AllowedVIPBenefitCodes) != 1 || presentation[0].AllowedVIPBenefitCodes[0] != ledger.VipBenefitCodeLeaderboardInvisible || len(presentation[1].AllowedVIPBenefitCodes) != 0 { + t.Fatalf("batch presentation must grant only the active VIP with its enabled setting: %+v", presentation) + } // 体验卡与付费 VIP 共用同一设置门禁；佩戴后 effective_benefits 仍保留资格，执行结果为 disabled。 const trialUserID int64 = 940002 diff --git a/services/wallet-service/internal/storage/mysql/host_salary_settlement.go b/services/wallet-service/internal/storage/mysql/host_salary_settlement.go index 1c8dd7ee..a3aa0dbf 100644 --- a/services/wallet-service/internal/storage/mysql/host_salary_settlement.go +++ b/services/wallet-service/internal/storage/mysql/host_salary_settlement.go @@ -530,7 +530,7 @@ func (r *Repository) queryBoundHostSalaryPolicy(ctx context.Context, q hostSalar if policy.PolicyType != ledger.HostPolicyTypeSalaryDiamond && policy.PolicyType != ledger.HostPolicyTypePointDiamond { return ledger.HostSalaryPolicy{}, false, xerr.New(xerr.Internal, "host policy type is invalid") } - levels, err := r.listHostSalaryPolicyLevels(ctx, q, policy.PolicyID) + levels, err := r.listHostSalaryPolicyLevels(ctx, q, policy.PolicyID, policy.PolicyVersion) if err != nil { return ledger.HostSalaryPolicy{}, false, err } @@ -589,14 +589,14 @@ func previousHostSalaryCycleKey(cycleKey string) (string, error) { return parsed.AddDate(0, -1, 0).Format("2006-01"), nil } -func (r *Repository) listHostSalaryPolicyLevels(ctx context.Context, q hostSalaryPolicyQuerier, policyID uint64) ([]ledger.HostSalaryPolicyLevel, error) { +func (r *Repository) listHostSalaryPolicyLevels(ctx context.Context, q hostSalaryPolicyQuerier, policyID uint64, policyVersion uint64) ([]ledger.HostSalaryPolicyLevel, error) { // 等级必须按 required_diamonds 升序返回，highestHostSalaryLevel 依赖这个顺序找到最高已达档位。 rows, err := q.QueryContext(ctx, ` SELECT level_no, required_diamonds, host_salary_usd_minor, host_coin_reward, agency_salary_usd_minor, status, sort_order FROM host_agency_salary_policy_levels - WHERE app_code = ? AND policy_id = ? AND status = 'active' - ORDER BY required_diamonds ASC, level_no ASC`, appcode.FromContext(ctx), policyID) + WHERE app_code = ? AND policy_id = ? AND policy_version = ? AND status = 'active' + ORDER BY required_diamonds ASC, level_no ASC`, appcode.FromContext(ctx), policyID, policyVersion) if err != nil { return nil, err } diff --git a/services/wallet-service/internal/storage/mysql/resource_entitlement_repository.go b/services/wallet-service/internal/storage/mysql/resource_entitlement_repository.go index 1cf68a23..c54712d2 100644 --- a/services/wallet-service/internal/storage/mysql/resource_entitlement_repository.go +++ b/services/wallet-service/internal/storage/mysql/resource_entitlement_repository.go @@ -523,6 +523,11 @@ func (r *Repository) BatchGetUserEquippedResources(ctx context.Context, query re if len(userIDs) == 0 { return []resourcedomain.UserEquippedResources{}, nil } + benefitCodes := normalizeVipBenefitCodes(query.VIPBenefitCodes) + // 展示批次只服务一页用户；显式限制矩阵规模，避免内部误用把动态 IN 查询扩大成无界扫描。 + if len(benefitCodes) > 16 || (len(benefitCodes) > 0 && len(userIDs) > 100) { + return nil, xerr.New(xerr.InvalidArgument, "vip benefit presentation batch is too large") + } resourceTypes, err := normalizeEquipableResourceTypes(query.ResourceTypes) if err != nil { @@ -590,12 +595,17 @@ func (r *Repository) BatchGetUserEquippedResources(ctx context.Context, query re if err := r.filterCurrentVipEquipmentBatch(ctx, userIDs, grouped, nowMs); err != nil { return nil, err } + allowedBenefits, err := r.batchAllowedVipBenefits(ctx, userIDs, benefitCodes, nowMs) + if err != nil { + return nil, err + } result := make([]resourcedomain.UserEquippedResources, 0, len(userIDs)) for _, userID := range userIDs { result = append(result, resourcedomain.UserEquippedResources{ - UserID: userID, - Resources: grouped[userID], + UserID: userID, + Resources: grouped[userID], + AllowedVIPBenefitCodes: allowedBenefits[userID], }) } return result, nil diff --git a/services/wallet-service/internal/storage/mysql/schema.go b/services/wallet-service/internal/storage/mysql/schema.go index 8396dd24..2424e311 100644 --- a/services/wallet-service/internal/storage/mysql/schema.go +++ b/services/wallet-service/internal/storage/mysql/schema.go @@ -63,7 +63,12 @@ func ensureHostSalarySettlementSchema(ctx context.Context, db *sql.DB) error { } if _, err := db.ExecContext(ctx, ` ALTER TABLE host_agency_salary_policies - ADD COLUMN policy_version BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '不可变政策版本，等于后台政策 ID' AFTER cycle_key`); err != nil && !isDuplicateColumnError(err) { + ADD COLUMN policy_version BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '不可变政策版本；首次等于后台政策 ID，修订后递增' AFTER cycle_key`); err != nil && !isDuplicateColumnError(err) { + return err + } + if _, err := db.ExecContext(ctx, ` + ALTER TABLE host_agency_salary_policy_levels + ADD COLUMN policy_version BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '所属不可变政策版本' AFTER policy_id`); err != nil && !isDuplicateColumnError(err) { return err } if _, err := db.ExecContext(ctx, ` @@ -266,6 +271,43 @@ func ensureHostSalarySettlementSchema(ctx context.Context, db *sql.DB) error { WHERE cycle_key = '' OR policy_version = 0`); err != nil { return err } + // 旧结构每个 policy_id 只有一份快照；小型等级表按复合主键点关联回填，不触碰钱包流水和工资记录。 + if _, err := db.ExecContext(ctx, ` + UPDATE host_agency_salary_policy_levels level + JOIN host_agency_salary_policies policy + ON policy.app_code = level.app_code + AND policy.policy_id = level.policy_id + SET level.policy_version = policy.policy_version + WHERE level.policy_version = 0`); err != nil { + return err + } + if columns, err := tablePrimaryKeyColumns(ctx, db, "host_agency_salary_policies"); err != nil { + return err + } else if strings.Join(columns, ",") != "app_code,policy_id,policy_version" { + if _, err := db.ExecContext(ctx, `ALTER TABLE host_agency_salary_policies DROP PRIMARY KEY, ADD PRIMARY KEY (app_code, policy_id, policy_version), ALGORITHM=INPLACE, LOCK=NONE`); err != nil { + return err + } + } + if columns, err := tablePrimaryKeyColumns(ctx, db, "host_agency_salary_policy_levels"); err != nil { + return err + } else if strings.Join(columns, ",") != "app_code,policy_id,policy_version,level_no" { + if _, err := db.ExecContext(ctx, `ALTER TABLE host_agency_salary_policy_levels DROP PRIMARY KEY, ADD PRIMARY KEY (app_code, policy_id, policy_version, level_no), ALGORITHM=INPLACE, LOCK=NONE`); err != nil { + return err + } + } + levelIndexColumns, err := tableIndexColumns(ctx, db, "host_agency_salary_policy_levels", "idx_host_agency_salary_policy_level_required") + if err != nil { + return err + } + if strings.Join(levelIndexColumns, ",") != "app_code,policy_id,policy_version,status,required_diamonds" { + if len(levelIndexColumns) > 0 { + if _, err := db.ExecContext(ctx, `ALTER TABLE host_agency_salary_policy_levels DROP INDEX idx_host_agency_salary_policy_level_required, ADD INDEX idx_host_agency_salary_policy_level_required (app_code, policy_id, policy_version, status, required_diamonds), ALGORITHM=INPLACE, LOCK=NONE`); err != nil { + return err + } + } else if _, err := db.ExecContext(ctx, `ALTER TABLE host_agency_salary_policy_levels ADD INDEX idx_host_agency_salary_policy_level_required (app_code, policy_id, policy_version, status, required_diamonds), ALGORITHM=INPLACE, LOCK=NONE`); err != nil { + return err + } + } if _, err := db.ExecContext(ctx, ` INSERT IGNORE INTO host_salary_policy_cycle_bindings ( app_code, region_id, cycle_key, policy_id, policy_version, source_cycle_key, created_at_ms diff --git a/services/wallet-service/internal/storage/mysql/vip_benefit_resource_repository.go b/services/wallet-service/internal/storage/mysql/vip_benefit_resource_repository.go index 5c1d2003..0cea9d18 100644 --- a/services/wallet-service/internal/storage/mysql/vip_benefit_resource_repository.go +++ b/services/wallet-service/internal/storage/mysql/vip_benefit_resource_repository.go @@ -685,57 +685,10 @@ func (r *Repository) filterCurrentVipEquipmentBatch(ctx context.Context, userIDs return tx.Commit() } - accessArgs := append([]any{appcode.FromContext(ctx)}, int64AnyArgs(userIDs)...) - accessArgs = append(accessArgs, nowMS, appcode.FromContext(ctx)) - accessArgs = append(accessArgs, int64AnyArgs(userIDs)...) - accessArgs = append(accessArgs, resourcedomain.TypeVIPTrialCard, ledger.VipTrialCardStatusActive, - nowMS, nowMS, nowMS, nowMS) - accessRows, err := tx.QueryContext(ctx, ` - SELECT membership.user_id, membership.level, 'paid' AS effective_source - FROM user_vip_memberships AS membership FORCE INDEX (PRIMARY) - WHERE membership.app_code = ? AND membership.user_id IN (`+placeholders(len(userIDs))+`) - AND membership.level > 0 AND membership.expires_at_ms > ? - UNION ALL - SELECT STRAIGHT_JOIN card.user_id, card.level, 'trial' AS effective_source - FROM user_resource_equipment AS equipment FORCE INDEX (PRIMARY) - JOIN user_vip_trial_cards AS card FORCE INDEX (uk_vip_trial_card_entitlement) ON card.app_code = equipment.app_code - AND card.user_id = equipment.user_id AND card.entitlement_id = equipment.entitlement_id - JOIN user_resource_entitlements AS entitlement ON entitlement.app_code = equipment.app_code - AND entitlement.user_id = equipment.user_id AND entitlement.entitlement_id = equipment.entitlement_id - JOIN resources AS resource ON resource.app_code = equipment.app_code AND resource.resource_id = equipment.resource_id - WHERE equipment.app_code = ? AND equipment.user_id IN (`+placeholders(len(userIDs))+`) - AND equipment.resource_type = ? AND card.status = ? - AND card.effective_at_ms <= ? AND card.expires_at_ms > ? - AND entitlement.status = 'active' AND entitlement.effective_at_ms <= ? - AND (entitlement.expires_at_ms = 0 OR entitlement.expires_at_ms > ?) - AND entitlement.remaining_quantity > 0 AND resource.status = 'active'`, accessArgs...) + levels, err := r.batchVipEffectiveLevelsFromSnapshot(ctx, tx, userIDs, nowMS) if err != nil { return err } - levels := make(map[int64]vipBatchEffectiveLevels, len(userIDs)) - for accessRows.Next() { - var userID int64 - var level int32 - var source string - if err := accessRows.Scan(&userID, &level, &source); err != nil { - _ = accessRows.Close() - return err - } - access := levels[userID] - if source == ledger.VipEffectiveSourceTrial { - access.Trial = level - } else { - access.Paid = level - } - levels[userID] = access - } - if err := accessRows.Err(); err != nil { - _ = accessRows.Close() - return err - } - if err := accessRows.Close(); err != nil { - return err - } filterVipEquipmentByAccess(grouped, historical, bindings, levels, program.TrialCardEnabled) return tx.Commit() } diff --git a/services/wallet-service/internal/storage/mysql/vip_program_repository.go b/services/wallet-service/internal/storage/mysql/vip_program_repository.go index 82919675..5bd420a5 100644 --- a/services/wallet-service/internal/storage/mysql/vip_program_repository.go +++ b/services/wallet-service/internal/storage/mysql/vip_program_repository.go @@ -4,12 +4,14 @@ import ( "context" "database/sql" "errors" + "fmt" + "strings" + "time" + "hyapp/pkg/appcode" "hyapp/pkg/xerr" "hyapp/services/wallet-service/internal/domain/ledger" resourcedomain "hyapp/services/wallet-service/internal/domain/resource" - "strings" - "time" ) type vipProgramScanner interface { @@ -79,6 +81,194 @@ func (r *Repository) GetVipState(ctx context.Context, userID int64) (ledger.VipS return r.getVipStateFromSnapshot(ctx, tx, userID, nowMS, config) } +// batchAllowedVipBenefits 在同一个只读快照内批量计算展示链路需要的最终权益。 +// 查询数量固定为 program、权益矩阵、用户 effective level、用户设置四次，且用户事实均以 +// (app_code,user_id) 索引收敛，不能退化成榜单每个用户各调一次 CheckVipBenefit。 +func (r *Repository) batchAllowedVipBenefits(ctx context.Context, userIDs []int64, rawBenefitCodes []string, nowMS int64) (map[int64][]string, error) { + result := make(map[int64][]string, len(userIDs)) + userIDs = compactPositiveInt64s(userIDs) + benefitCodes := normalizeVipBenefitCodes(rawBenefitCodes) + if len(userIDs) == 0 || len(benefitCodes) == 0 { + return result, nil + } + + tx, err := r.db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true, Isolation: sql.LevelRepeatableRead}) + if err != nil { + return nil, err + } + defer func() { _ = tx.Rollback() }() + program, err := r.requireVipProgramConfig(ctx, tx) + if err != nil { + return nil, err + } + if program.Status != ledger.VipStatusActive { + return result, tx.Commit() + } + + type benefitAccess struct { + TrialEnabled bool + } + benefits := make(map[string]benefitAccess, len(benefitCodes)*int(program.LevelCount)) + benefitArgs := append([]any{appcode.FromContext(ctx)}, stringAnyArgs(benefitCodes)...) + benefitArgs = append(benefitArgs, ledger.VipStatusActive) + benefitRows, err := tx.QueryContext(ctx, ` + SELECT level, benefit_code, trial_enabled + FROM vip_level_benefits FORCE INDEX (idx_vip_benefit_code) + WHERE app_code = ? AND benefit_code IN (`+placeholders(len(benefitCodes))+`) + AND status = ?`, benefitArgs...) + if err != nil { + return nil, err + } + for benefitRows.Next() { + var level int32 + var code string + var trialEnabled bool + if err := benefitRows.Scan(&level, &code, &trialEnabled); err != nil { + _ = benefitRows.Close() + return nil, err + } + benefits[fmt.Sprintf("%d|%s", level, strings.ToLower(strings.TrimSpace(code)))] = benefitAccess{TrialEnabled: trialEnabled} + } + if err := benefitRows.Err(); err != nil { + _ = benefitRows.Close() + return nil, err + } + if err := benefitRows.Close(); err != nil { + return nil, err + } + + levels, err := r.batchVipEffectiveLevelsFromSnapshot(ctx, tx, userIDs, nowMS) + if err != nil { + return nil, err + } + settings := make(map[int64]ledger.VipUserSettings, len(userIDs)) + for _, userID := range userIDs { + settings[userID] = ledger.DefaultVipUserSettings(appcode.FromContext(ctx), userID) + } + settingArgs := append([]any{appcode.FromContext(ctx)}, int64AnyArgs(userIDs)...) + settingRows, err := tx.QueryContext(ctx, ` + SELECT user_id, room_entry_notice_enabled, online_global_notice_enabled, + hide_profile_data_enabled, anonymous_profile_visit_enabled, + leaderboard_invisible_enabled, updated_at_ms + FROM user_vip_settings FORCE INDEX (PRIMARY) + WHERE app_code = ? AND user_id IN (`+placeholders(len(userIDs))+`)`, settingArgs...) + if err != nil { + return nil, err + } + for settingRows.Next() { + var value ledger.VipUserSettings + value.AppCode = appcode.FromContext(ctx) + if err := settingRows.Scan( + &value.UserID, &value.RoomEntryNoticeEnabled, &value.OnlineGlobalNoticeEnabled, + &value.HideProfileDataEnabled, &value.AnonymousProfileVisitEnabled, + &value.LeaderboardInvisibleEnabled, &value.UpdatedAtMS, + ); err != nil { + _ = settingRows.Close() + return nil, err + } + settings[value.UserID] = value + } + if err := settingRows.Err(); err != nil { + _ = settingRows.Close() + return nil, err + } + if err := settingRows.Close(); err != nil { + return nil, err + } + + for _, userID := range userIDs { + access := levels[userID] + effectiveLevel := access.Paid + effectiveSource := ledger.VipEffectiveSourcePaid + if program.TrialCardEnabled && access.Trial > 0 { + effectiveLevel = access.Trial + effectiveSource = ledger.VipEffectiveSourceTrial + } + if effectiveLevel <= 0 { + continue + } + for _, code := range benefitCodes { + benefit, exists := benefits[fmt.Sprintf("%d|%s", effectiveLevel, code)] + if !exists || !settings[userID].AllowsBenefitSetting(code) { + continue + } + // 体验卡只能执行该等级显式允许 trial 的权益；金币返现保持付费会员硬边界。 + if effectiveSource == ledger.VipEffectiveSourceTrial && (!benefit.TrialEnabled || code == ledger.VipBenefitCodeDailyCoinRebate) { + continue + } + result[userID] = append(result[userID], code) + } + } + return result, tx.Commit() +} + +func normalizeVipBenefitCodes(values []string) []string { + seen := make(map[string]struct{}, len(values)) + result := make([]string, 0, len(values)) + for _, value := range values { + code := strings.ToLower(strings.TrimSpace(value)) + if code == "" { + continue + } + if _, exists := seen[code]; exists { + continue + } + seen[code] = struct{}{} + result = append(result, code) + } + return result +} + +// batchVipEffectiveLevelsFromSnapshot 复用 GetVipState 的 paid/trial 优先级事实，但按一批用户 +// 只读取一次 membership 和一次已佩戴体验卡集合。所有时间条件与单用户读取保持一致。 +func (r *Repository) batchVipEffectiveLevelsFromSnapshot(ctx context.Context, querier vipSnapshotQuerier, userIDs []int64, nowMS int64) (map[int64]vipBatchEffectiveLevels, error) { + accessArgs := append([]any{appcode.FromContext(ctx)}, int64AnyArgs(userIDs)...) + accessArgs = append(accessArgs, nowMS, appcode.FromContext(ctx)) + accessArgs = append(accessArgs, int64AnyArgs(userIDs)...) + accessArgs = append(accessArgs, resourcedomain.TypeVIPTrialCard, ledger.VipTrialCardStatusActive, + nowMS, nowMS, nowMS, nowMS) + rows, err := querier.QueryContext(ctx, ` + SELECT membership.user_id, membership.level, 'paid' AS effective_source + FROM user_vip_memberships AS membership FORCE INDEX (PRIMARY) + WHERE membership.app_code = ? AND membership.user_id IN (`+placeholders(len(userIDs))+`) + AND membership.level > 0 AND membership.expires_at_ms > ? + UNION ALL + SELECT STRAIGHT_JOIN card.user_id, card.level, 'trial' AS effective_source + FROM user_resource_equipment AS equipment FORCE INDEX (PRIMARY) + JOIN user_vip_trial_cards AS card FORCE INDEX (uk_vip_trial_card_entitlement) ON card.app_code = equipment.app_code + AND card.user_id = equipment.user_id AND card.entitlement_id = equipment.entitlement_id + JOIN user_resource_entitlements AS entitlement ON entitlement.app_code = equipment.app_code + AND entitlement.user_id = equipment.user_id AND entitlement.entitlement_id = equipment.entitlement_id + JOIN resources AS resource ON resource.app_code = equipment.app_code AND resource.resource_id = equipment.resource_id + WHERE equipment.app_code = ? AND equipment.user_id IN (`+placeholders(len(userIDs))+`) + AND equipment.resource_type = ? AND card.status = ? + AND card.effective_at_ms <= ? AND card.expires_at_ms > ? + AND entitlement.status = 'active' AND entitlement.effective_at_ms <= ? + AND (entitlement.expires_at_ms = 0 OR entitlement.expires_at_ms > ?) + AND entitlement.remaining_quantity > 0 AND resource.status = 'active'`, accessArgs...) + if err != nil { + return nil, err + } + defer rows.Close() + levels := make(map[int64]vipBatchEffectiveLevels, len(userIDs)) + for rows.Next() { + var userID int64 + var level int32 + var source string + if err := rows.Scan(&userID, &level, &source); err != nil { + return nil, err + } + access := levels[userID] + if source == ledger.VipEffectiveSourceTrial { + access.Trial = level + } else { + access.Paid = level + } + levels[userID] = access + } + return levels, rows.Err() +} + // requireVipProgramConfig 从调用者指定的快照读取 App 配置，并把缺迁移/错 app_code 明确映射为业务错误。 func (r *Repository) requireVipProgramConfig(ctx context.Context, querier vipSnapshotQuerier) (ledger.VipProgramConfig, error) { config, err := r.queryVipProgramConfig(ctx, querier.QueryRowContext(ctx, vipProgramSelectSQL(), appcode.FromContext(ctx))) diff --git a/services/wallet-service/internal/testutil/mysqltest/mysqltest.go b/services/wallet-service/internal/testutil/mysqltest/mysqltest.go index 23599857..f5f0be6a 100644 --- a/services/wallet-service/internal/testutil/mysqltest/mysqltest.go +++ b/services/wallet-service/internal/testutil/mysqltest/mysqltest.go @@ -1073,7 +1073,7 @@ func (r *Repository) SetHostSalaryPolicyForApp(appCode string, policy ledger.Hos if err != nil { r.t.Fatalf("seed host salary policy failed: %v", err) } - if _, err := r.schema.DB.ExecContext(context.Background(), `DELETE FROM host_agency_salary_policy_levels WHERE app_code = ? AND policy_id = ?`, appCode, policy.PolicyID); err != nil { + if _, err := r.schema.DB.ExecContext(context.Background(), `DELETE FROM host_agency_salary_policy_levels WHERE app_code = ? AND policy_id = ? AND policy_version = ?`, appCode, policy.PolicyID, policy.PolicyVersion); err != nil { r.t.Fatalf("clear host salary policy levels failed: %v", err) } for _, level := range policy.Levels { @@ -1082,10 +1082,10 @@ func (r *Repository) SetHostSalaryPolicyForApp(appCode string, policy ledger.Hos } if _, err := r.schema.DB.ExecContext(context.Background(), ` INSERT INTO host_agency_salary_policy_levels ( - app_code, policy_id, level_no, required_diamonds, host_salary_usd_minor, + app_code, policy_id, policy_version, level_no, required_diamonds, host_salary_usd_minor, host_coin_reward, agency_salary_usd_minor, status, sort_order, created_at_ms, updated_at_ms - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, - appCode, policy.PolicyID, level.LevelNo, level.RequiredDiamonds, level.HostSalaryUSDMinor, + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + appCode, policy.PolicyID, policy.PolicyVersion, level.LevelNo, level.RequiredDiamonds, level.HostSalaryUSDMinor, level.HostCoinReward, level.AgencySalaryUSDMinor, level.Status, level.SortOrder, nowMs, nowMs, ); err != nil { r.t.Fatalf("seed host salary policy level %d failed: %v", level.LevelNo, err) diff --git a/services/wallet-service/internal/transport/grpc/resource.go b/services/wallet-service/internal/transport/grpc/resource.go index a6ba58f0..39b28ded 100644 --- a/services/wallet-service/internal/transport/grpc/resource.go +++ b/services/wallet-service/internal/transport/grpc/resource.go @@ -415,9 +415,10 @@ func (s *Server) UnequipUserResource(ctx context.Context, req *walletv1.UnequipU func (s *Server) BatchGetUserEquippedResources(ctx context.Context, req *walletv1.BatchGetUserEquippedResourcesRequest) (*walletv1.BatchGetUserEquippedResourcesResponse, error) { items, err := s.svc.BatchGetUserEquippedResources(ctx, resourcedomain.BatchGetUserEquippedResourcesQuery{ - AppCode: req.GetAppCode(), - UserIDs: req.GetUserIds(), - ResourceTypes: req.GetResourceTypes(), + AppCode: req.GetAppCode(), + UserIDs: req.GetUserIds(), + ResourceTypes: req.GetResourceTypes(), + VIPBenefitCodes: req.GetVipBenefitCodes(), }) if err != nil { return nil, xerr.ToGRPCError(err) @@ -425,8 +426,9 @@ func (s *Server) BatchGetUserEquippedResources(ctx context.Context, req *walletv resp := &walletv1.BatchGetUserEquippedResourcesResponse{Users: make([]*walletv1.UserEquippedResources, 0, len(items))} for _, item := range items { user := &walletv1.UserEquippedResources{ - UserId: item.UserID, - Resources: make([]*walletv1.UserResourceEntitlement, 0, len(item.Resources)), + UserId: item.UserID, + Resources: make([]*walletv1.UserResourceEntitlement, 0, len(item.Resources)), + AllowedVipBenefitCodes: append([]string(nil), item.AllowedVIPBenefitCodes...), } for _, resource := range item.Resources { user.Resources = append(user.Resources, userResourceToProto(resource)) diff --git a/tools/databi-real-flow/main.go b/tools/databi-real-flow/main.go index 20d393f3..68a00bfd 100644 --- a/tools/databi-real-flow/main.go +++ b/tools/databi-real-flow/main.go @@ -238,16 +238,16 @@ func (r *runner) ensureSalaryConfig(ctx context.Context) error { } _, err = r.walletDB.ExecContext(ctx, ` INSERT INTO host_agency_salary_policy_levels ( - app_code, policy_id, level_no, required_diamonds, host_salary_usd_minor, host_coin_reward, + app_code, policy_id, policy_version, level_no, required_diamonds, host_salary_usd_minor, host_coin_reward, agency_salary_usd_minor, status, sort_order, created_at_ms, updated_at_ms - ) VALUES (?, ?, 1, 100, 150, 0, 50, 'active', 1, ?, ?) + ) VALUES (?, ?, ?, 1, 100, 150, 0, 50, 'active', 1, ?, ?) ON DUPLICATE KEY UPDATE required_diamonds = VALUES(required_diamonds), host_salary_usd_minor = VALUES(host_salary_usd_minor), agency_salary_usd_minor = VALUES(agency_salary_usd_minor), status = 'active', updated_at_ms = VALUES(updated_at_ms)`, - r.cfg.appCode, policyID, now, now) + r.cfg.appCode, policyID, policyID, now, now) if err != nil { return fmt.Errorf("seed salary policy level: %w", err) }