diff --git a/server/admin/cmd/server/main.go b/server/admin/cmd/server/main.go index 1d1a3f2d..27966077 100644 --- a/server/admin/cmd/server/main.go +++ b/server/admin/cmd/server/main.go @@ -293,7 +293,7 @@ func main() { Resource: resourcemodule.New(walletclient.NewGRPC(walletConn), store, userDB, cfg.WalletService.RequestTimeout, auditHandler), RoomAdmin: roomadminmodule.New(userDB, roomClient, robotClient, auditHandler), RoomRocket: roomrocketmodule.New(roomClient, auditHandler), - RoomTurnoverReward: roomturnoverrewardmodule.New(activityclient.NewGRPC(activityConn), auditHandler), + RoomTurnoverReward: roomturnoverrewardmodule.New(activityclient.NewGRPC(activityConn), userDB, auditHandler), Search: searchmodule.New(store), SevenDayCheckIn: sevendaycheckinmodule.New(activityclient.NewGRPC(activityConn), userDB, auditHandler), TeamSalaryPolicy: teamsalarypolicymodule.New(store, auditHandler), diff --git a/server/admin/internal/modules/roomadmin/robot_room.go b/server/admin/internal/modules/roomadmin/robot_room.go index 600a32d4..60d46b24 100644 --- a/server/admin/internal/modules/roomadmin/robot_room.go +++ b/server/admin/internal/modules/roomadmin/robot_room.go @@ -136,7 +136,12 @@ func (s *Service) CreateRobotRoom(ctx context.Context, req createRobotRoomReques if s.roomClient == nil { return RobotRoom{}, fmt.Errorf("room service client is not configured") } - if _, err := normalizeCreateRobotRoomRequest(req); err != nil { + normalized, err := normalizeCreateRobotRoomRequest(req) + if err != nil { + return RobotRoom{}, err + } + req = normalized + if err := s.applyOwnerProfileToRobotRoomRequest(ctx, &req); err != nil { return RobotRoom{}, err } created, err := s.roomClient.CreateRobotRoom(ctx, roomclient.CreateRobotRoomRequest{ @@ -168,6 +173,40 @@ func (s *Service) CreateRobotRoom(ctx context.Context, req createRobotRoomReques return items[0], nil } +func (s *Service) applyOwnerProfileToRobotRoomRequest(ctx context.Context, req *createRobotRoomRequest) error { + if s.userDB == nil { + return fmt.Errorf("user database is not configured") + } + owners, err := s.queryRoomOwners(ctx, []int64{req.OwnerRobotUserID}) + if err != nil { + return err + } + owner, ok := owners[req.OwnerRobotUserID] + if !ok { + return fmt.Errorf("%w: 房主机器人资料不存在", ErrInvalidArgument) + } + roomName, roomAvatar, err := robotRoomProfileFromOwner(owner) + if err != nil { + return err + } + // 机器人房间对外展示必须跟随房主机器人资料，避免后台表单或脚本传入的自定义值让房间卡片和房主身份不一致。 + req.RoomName = roomName + req.RoomAvatar = roomAvatar + return nil +} + +func robotRoomProfileFromOwner(owner RoomOwner) (string, string, error) { + name := strings.TrimSpace(owner.Username) + avatar := strings.TrimSpace(owner.Avatar) + if name == "" { + return "", "", fmt.Errorf("%w: 房主机器人名称不能为空", ErrInvalidArgument) + } + if avatar == "" { + return "", "", fmt.Errorf("%w: 房主机器人头像不能为空", ErrInvalidArgument) + } + return name, avatar, nil +} + func (s *Service) SetRobotRoomStatus(ctx context.Context, roomID string, status string, actor shared.Actor) (RobotRoom, error) { if s.roomClient == nil { return RobotRoom{}, fmt.Errorf("room service client is not configured") diff --git a/server/admin/internal/modules/roomadmin/service_test.go b/server/admin/internal/modules/roomadmin/service_test.go index 7e1364e3..544d04a2 100644 --- a/server/admin/internal/modules/roomadmin/service_test.go +++ b/server/admin/internal/modules/roomadmin/service_test.go @@ -1,6 +1,9 @@ package roomadmin -import "testing" +import ( + "errors" + "testing" +) func TestNormalizeRoomListSortKeepsOnlyContributionSort(t *testing.T) { tests := []struct { @@ -39,3 +42,35 @@ func TestNormalizeRoomListSortKeepsOnlyContributionSort(t *testing.T) { }) } } + +func TestRobotRoomProfileFromOwnerUsesOwnerNameAndAvatar(t *testing.T) { + name, avatar, err := robotRoomProfileFromOwner(RoomOwner{ + Username: " Robot Host ", + Avatar: " https://cdn.example.com/robot.png ", + }) + if err != nil { + t.Fatalf("robot room owner profile should be valid: %v", err) + } + if name != "Robot Host" || avatar != "https://cdn.example.com/robot.png" { + t.Fatalf("owner profile mismatch: name=%q avatar=%q", name, avatar) + } +} + +func TestRobotRoomProfileFromOwnerRequiresNameAndAvatar(t *testing.T) { + tests := []struct { + name string + owner RoomOwner + }{ + {name: "missing_name", owner: RoomOwner{Avatar: "https://cdn.example.com/robot.png"}}, + {name: "missing_avatar", owner: RoomOwner{Username: "Robot Host"}}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + _, _, err := robotRoomProfileFromOwner(test.owner) + if !errors.Is(err, ErrInvalidArgument) { + t.Fatalf("expected ErrInvalidArgument, got %v", err) + } + }) + } +} diff --git a/server/admin/internal/modules/roomturnoverreward/handler.go b/server/admin/internal/modules/roomturnoverreward/handler.go index fce97d1b..aef970e1 100644 --- a/server/admin/internal/modules/roomturnoverreward/handler.go +++ b/server/admin/internal/modules/roomturnoverreward/handler.go @@ -1,6 +1,9 @@ package roomturnoverreward import ( + "context" + "database/sql" + "strconv" "strings" "time" @@ -16,11 +19,12 @@ import ( type Handler struct { activity activityclient.Client + userDB *sql.DB audit shared.OperationLogger } -func New(activity activityclient.Client, audit shared.OperationLogger) *Handler { - return &Handler{activity: activity, audit: audit} +func New(activity activityclient.Client, userDB *sql.DB, audit shared.OperationLogger) *Handler { + return &Handler{activity: activity, userDB: userDB, audit: audit} } type configRequest struct { @@ -50,23 +54,31 @@ type tierDTO struct { } type settlementDTO struct { - SettlementID string `json:"settlement_id"` - RoomID string `json:"room_id"` - OwnerUserID int64 `json:"owner_user_id,string"` - PeriodStartMS int64 `json:"period_start_ms"` - PeriodEndMS int64 `json:"period_end_ms"` - CoinSpent int64 `json:"coin_spent"` - TierID int64 `json:"tier_id"` - TierCode string `json:"tier_code"` - ThresholdCoinSpent int64 `json:"threshold_coin_spent"` - RewardCoinAmount int64 `json:"reward_coin_amount"` - Status string `json:"status"` - WalletCommandID string `json:"wallet_command_id"` - WalletTransactionID string `json:"wallet_transaction_id"` - FailureReason string `json:"failure_reason"` - SettledAtMS int64 `json:"settled_at_ms"` - CreatedAtMS int64 `json:"created_at_ms"` - UpdatedAtMS int64 `json:"updated_at_ms"` + SettlementID string `json:"settlement_id"` + RoomID string `json:"room_id"` + OwnerUserID int64 `json:"owner_user_id,string"` + OwnerUser *userDTO `json:"owner_user,omitempty"` + PeriodStartMS int64 `json:"period_start_ms"` + PeriodEndMS int64 `json:"period_end_ms"` + CoinSpent int64 `json:"coin_spent"` + TierID int64 `json:"tier_id"` + TierCode string `json:"tier_code"` + ThresholdCoinSpent int64 `json:"threshold_coin_spent"` + RewardCoinAmount int64 `json:"reward_coin_amount"` + Status string `json:"status"` + WalletCommandID string `json:"wallet_command_id"` + WalletTransactionID string `json:"wallet_transaction_id"` + FailureReason string `json:"failure_reason"` + SettledAtMS int64 `json:"settled_at_ms"` + CreatedAtMS int64 `json:"created_at_ms"` + UpdatedAtMS int64 `json:"updated_at_ms"` +} + +type userDTO struct { + UserID int64 `json:"user_id,string"` + DisplayUserID string `json:"display_user_id"` + Username string `json:"username"` + Avatar string `json:"avatar"` } func (h *Handler) GetConfig(c *gin.Context) { @@ -117,12 +129,26 @@ func (h *Handler) UpdateConfig(c *gin.Context) { func (h *Handler) ListSettlements(c *gin.Context) { // 列表筛选沿用通用 ListOptions：status 过滤结算状态，keyword 用作 room_id 精确查询，避免后台接口引入模糊扫表。 options := shared.ListOptions(c) + ownerKeyword := strings.TrimSpace(firstQuery(c, "owner_user_keyword", "ownerUserKeyword")) + ownerUserID, ownerMatched, ownerOK := h.resolveOwnerUserID(c.Request.Context(), ownerKeyword) + if !ownerOK { + response.ServerError(c, "查询房主信息失败") + return + } + if !ownerMatched { + ownerUserID = queryInt64(c, "owner_user_id", "ownerUserId") + } + if !ownerMatched && ownerKeyword != "" { + response.OK(c, response.Page{Items: []settlementDTO{}, Page: options.Page, PageSize: options.PageSize, Total: 0}) + return + } resp, err := h.activity.ListRoomTurnoverRewardSettlements(c.Request.Context(), &activityv1.ListRoomTurnoverRewardSettlementsRequest{ - Meta: h.meta(c), - Status: strings.TrimSpace(options.Status), - RoomId: strings.TrimSpace(options.Keyword), - Page: int32(options.Page), - PageSize: int32(options.PageSize), + Meta: h.meta(c), + Status: strings.TrimSpace(options.Status), + RoomId: strings.TrimSpace(options.Keyword), + OwnerUserId: ownerUserID, + Page: int32(options.Page), + PageSize: int32(options.PageSize), }) if err != nil { response.ServerError(c, "获取房间流水奖励结算记录失败") @@ -132,6 +158,11 @@ func (h *Handler) ListSettlements(c *gin.Context) { for _, settlement := range resp.GetSettlements() { items = append(items, settlementFromProto(settlement)) } + // activity-service 返回的是结算事实快照；房主昵称、头像和短号只服务后台展示，所以在 admin-server 按当前页 owner_user_id 批量补齐。 + if err := h.fillOwnerUsers(c.Request.Context(), items); err != nil { + response.ServerError(c, "补全房主信息失败") + return + } response.OK(c, response.Page{Items: items, Page: options.Page, PageSize: options.PageSize, Total: resp.GetTotal()}) } @@ -166,6 +197,100 @@ func (h *Handler) meta(c *gin.Context) *activityv1.RequestMeta { } } +func (h *Handler) resolveOwnerUserID(ctx context.Context, keyword string) (int64, bool, bool) { + if keyword == "" { + return 0, false, true + } + if h.userDB == nil { + // userDB 缺失时只允许长 ID 精确筛选，避免把短号或昵称误当成 settlement.owner_user_id 查询。 + if parsed, err := strconv.ParseInt(keyword, 10, 64); err == nil && parsed > 0 { + return parsed, true, true + } + return 0, false, true + } + appCode := appctx.FromContext(ctx) + nowMs := time.Now().UnixMilli() + matchSQL, matchArgs := shared.UserIdentityExactSQL("u", "u.user_id", keyword, nowMs) + orderSQL, orderArgs := shared.UserIdentityExactOrderSQL("u", "u.user_id", keyword, nowMs) + args := append([]any{appCode}, matchArgs...) + args = append(args, orderArgs...) + // 表头筛选先把运营输入解析成唯一房主 user_id，再交给 activity-service 的 owner_user_id 条件，保证分页 total 和列表数据同源。 + rows, err := h.userDB.QueryContext(ctx, ` + SELECT u.user_id + FROM users u + WHERE u.app_code = ? AND `+matchSQL+` + ORDER BY + `+orderSQL+`, + u.user_id DESC + LIMIT 1`, + args..., + ) + if err != nil { + return 0, false, false + } + defer rows.Close() + if !rows.Next() { + return 0, false, rows.Err() == nil + } + var userID int64 + if err := rows.Scan(&userID); err != nil { + return 0, false, false + } + return userID, true, rows.Err() == nil +} + +func (h *Handler) fillOwnerUsers(ctx context.Context, settlements []settlementDTO) error { + if len(settlements) == 0 { + return nil + } + ids := collectOwnerUserIDs(settlements) + if len(ids) == 0 { + return nil + } + users, err := h.loadOwnerUsers(ctx, ids) + if err != nil { + return err + } + for index := range settlements { + if user, ok := users[settlements[index].OwnerUserID]; ok { + settlements[index].OwnerUser = &user + continue + } + // 历史结算即使用户资料缺失，也必须保留收款人长 ID，方便后台继续追踪发奖对象。 + settlements[index].OwnerUser = &userDTO{UserID: settlements[index].OwnerUserID} + } + return nil +} + +func (h *Handler) loadOwnerUsers(ctx context.Context, ids []int64) (map[int64]userDTO, error) { + if h.userDB == nil || len(ids) == 0 { + return map[int64]userDTO{}, nil + } + args := append([]any{appctx.FromContext(ctx)}, int64Args(ids)...) + rows, err := h.userDB.QueryContext(ctx, ` + SELECT user_id, current_display_user_id, COALESCE(username, ''), COALESCE(avatar, '') + FROM users + WHERE app_code = ? AND user_id IN (`+placeholders(len(ids))+`)`, + args..., + ) + if err != nil { + return nil, err + } + defer rows.Close() + users := make(map[int64]userDTO, len(ids)) + for rows.Next() { + var user userDTO + if err := rows.Scan(&user.UserID, &user.DisplayUserID, &user.Username, &user.Avatar); err != nil { + return nil, err + } + users[user.UserID] = user + } + if err := rows.Err(); err != nil { + return nil, err + } + return users, nil +} + func configFromProto(config *activityv1.RoomTurnoverRewardConfig) configDTO { if config == nil { return configDTO{Tiers: []tierDTO{}} @@ -184,6 +309,62 @@ func configFromProto(config *activityv1.RoomTurnoverRewardConfig) configDTO { } } +func collectOwnerUserIDs(settlements []settlementDTO) []int64 { + seen := make(map[int64]struct{}, len(settlements)) + ids := make([]int64, 0, len(settlements)) + for _, settlement := range settlements { + if settlement.OwnerUserID <= 0 { + continue + } + if _, ok := seen[settlement.OwnerUserID]; ok { + continue + } + seen[settlement.OwnerUserID] = struct{}{} + ids = append(ids, settlement.OwnerUserID) + } + return ids +} + +func int64Args(ids []int64) []any { + args := make([]any, 0, len(ids)) + for _, id := range ids { + args = append(args, id) + } + return args +} + +func placeholders(count int) string { + if count <= 0 { + return "" + } + parts := make([]string, count) + for i := range parts { + parts[i] = "?" + } + return strings.Join(parts, ",") +} + +func queryInt64(c *gin.Context, keys ...string) int64 { + value := strings.TrimSpace(firstQuery(c, keys...)) + if value == "" { + return 0 + } + parsed, err := strconv.ParseInt(value, 10, 64) + if err != nil || parsed < 0 { + return 0 + } + return parsed +} + +func firstQuery(c *gin.Context, keys ...string) string { + for _, key := range keys { + if value := c.Query(key); value != "" { + return value + } + } + return "" +} + func tierFromProto(tier *activityv1.RoomTurnoverRewardTier) tierDTO { if tier == nil { return tierDTO{} diff --git a/services/activity-service/internal/domain/agencyopening/agency_opening.go b/services/activity-service/internal/domain/agencyopening/agency_opening.go index fcbd1e93..fdb46b91 100644 --- a/services/activity-service/internal/domain/agencyopening/agency_opening.go +++ b/services/activity-service/internal/domain/agencyopening/agency_opening.go @@ -21,7 +21,7 @@ const ( GrantSourceAgencyOpening = "agency_opening" ) -// Reward 是代理开业流水档位；RankNo 只保留为档位展示顺序和旧字段兼容，结算命中只看 ThresholdCoinSpent。 +// Reward 是代理开业贡献档位；RankNo 只保留为档位展示顺序和旧字段兼容，结算命中只看 ThresholdCoinSpent。 type Reward struct { AppCode string CycleID string @@ -124,6 +124,7 @@ type Qualification struct { } // GiftEvent is the room-service gift fact projected into the applicant room owner's opening revenue. +// CoinSpent keeps the legacy field name but stores RoomGiftSent.gift_value/HeatValue. type GiftEvent struct { EventID string RoomID string diff --git a/services/activity-service/internal/domain/roomturnoverreward/room_turnover_reward.go b/services/activity-service/internal/domain/roomturnoverreward/room_turnover_reward.go index 04cfab4b..7e0e5e52 100644 --- a/services/activity-service/internal/domain/roomturnoverreward/room_turnover_reward.go +++ b/services/activity-service/internal/domain/roomturnoverreward/room_turnover_reward.go @@ -13,7 +13,7 @@ const ( EventStatusSkipped = "skipped" ) -// Tier 是后台配置的房间流水奖励档位，结算只发命中的最高有效档。 +// Tier 是后台配置的房间贡献奖励档位，结算只发命中的最高有效档。 type Tier struct { TierID int64 TierCode string @@ -36,7 +36,7 @@ type Config struct { UpdatedAtMS int64 } -// Progress 是单房间单 UTC 周期的金币流水聚合。 +// Progress 是单房间单贡献周期的 HeatValue 聚合；CoinSpent 是历史字段名，运行值为房间贡献值。 type Progress struct { AppCode string RoomID string @@ -48,7 +48,7 @@ type Progress struct { UpdatedAtMS int64 } -// RoomGiftEvent 是 room-service RoomGiftSent 事实映射后的最小聚合输入。 +// RoomGiftEvent 是 room-service RoomGiftSent 事实映射后的最小聚合输入；CoinSpent 字段承载 gift_value/HeatValue。 type RoomGiftEvent struct { EventID string EventType string @@ -64,7 +64,7 @@ type EventResult struct { Status string } -// Settlement 是某个房间单个 UTC 周期的奖励发放事实。 +// Settlement 是某个房间单个贡献周期的奖励发放事实；CoinSpent 字段承载周期 HeatValue。 type Settlement struct { SettlementID string AppCode string diff --git a/services/activity-service/internal/service/agencyopening/service.go b/services/activity-service/internal/service/agencyopening/service.go index d334161b..6d653448 100644 --- a/services/activity-service/internal/service/agencyopening/service.go +++ b/services/activity-service/internal/service/agencyopening/service.go @@ -199,7 +199,8 @@ func (s *Service) HandleRoomEvent(ctx context.Context, envelope *roomeventsv1.Ev return 0, err } roomID := strings.TrimSpace(envelope.GetRoomId()) - if gift.GetIsRobotGift() || roomID == "" || gift.GetCoinSpent() <= 0 { + contributionValue := gift.GetGiftValue() + if gift.GetIsRobotGift() || roomID == "" || contributionValue <= 0 { return 0, nil } eventCtx := appcode.WithContext(ctx, envelope.GetAppCode()) @@ -212,8 +213,9 @@ func (s *Service) HandleRoomEvent(ctx context.Context, envelope *roomeventsv1.Ev RoomID: roomID, RoomOwnerUserID: ownerUserID, TargetUserID: gift.GetTargetUserId(), - CoinSpent: gift.GetCoinSpent(), - OccurredAtMS: envelope.GetOccurredAtMs(), + // 开业流水只消费 wallet 结算后的 HeatValue，也就是 RoomGiftSent.gift_value；原始扣费 coin_spent 不再参与奖励达标。 + CoinSpent: contributionValue, + OccurredAtMS: envelope.GetOccurredAtMs(), }) } diff --git a/services/activity-service/internal/service/agencyopening/service_test.go b/services/activity-service/internal/service/agencyopening/service_test.go index c7fe39a7..0843c152 100644 --- a/services/activity-service/internal/service/agencyopening/service_test.go +++ b/services/activity-service/internal/service/agencyopening/service_test.go @@ -81,6 +81,7 @@ func TestHandleRoomEventScoresApplicantRoomOwner(t *testing.T) { body, err := proto.Marshal(&roomeventsv1.RoomGiftSent{ SenderUserId: 9, TargetUserId: 88, + GiftValue: 900, CoinSpent: 1200, }) if err != nil { @@ -103,7 +104,7 @@ func TestHandleRoomEventScoresApplicantRoomOwner(t *testing.T) { if room.roomID != "room-1" { t.Fatalf("expected room lookup by event room_id, got %q", room.roomID) } - if repo.consumed.RoomOwnerUserID != 42 || repo.consumed.TargetUserID != 88 || repo.consumed.CoinSpent != 1200 { + if repo.consumed.RoomOwnerUserID != 42 || repo.consumed.TargetUserID != 88 || repo.consumed.CoinSpent != 900 { t.Fatalf("unexpected consumed event: %+v", repo.consumed) } } diff --git a/services/activity-service/internal/service/roomturnoverreward/service.go b/services/activity-service/internal/service/roomturnoverreward/service.go index 6a8cc1f2..521d989a 100644 --- a/services/activity-service/internal/service/roomturnoverreward/service.go +++ b/services/activity-service/internal/service/roomturnoverreward/service.go @@ -45,7 +45,7 @@ type RoomQueryClient interface { AdminGetRoom(ctx context.Context, req *roomv1.AdminGetRoomRequest, opts ...grpc.CallOption) (*roomv1.AdminGetRoomResponse, error) } -// Service 承载房间流水奖励 App 查询、后台配置、事件聚合和每周结算。 +// Service 承载房间贡献奖励 App 查询、后台配置、事件聚合和每周结算。 type Service struct { repository Repository wallet WalletClient @@ -114,7 +114,7 @@ func (s *Service) GetStatus(ctx context.Context, userID int64, roomID string, ow config.Tiers = sortTiersForDisplay(config.Tiers) result.Config = config } else { - // 未配置不是异常：后台首次进入或活动未上线时，H5 仍需要周期时间和空档位来渲染默认态。 + // 未配置不是异常：后台首次进入或活动未上线时，H5 仍需要贡献周期时间和空档位来渲染默认态。 result.Config = domain.Config{AppCode: appcode.FromContext(ctx)} } if result.RoomID == "" { @@ -141,7 +141,7 @@ func (s *Service) GetStatus(ctx context.Context, userID int64, roomID string, ow return result, nil } -// HandleRoomEvent 把 room-service 已提交送礼事实聚合到对应 UTC 周期房间流水。 +// HandleRoomEvent 把 room-service 已提交送礼事实聚合到对应贡献周期房间贡献值。 func (s *Service) HandleRoomEvent(ctx context.Context, envelope *roomeventsv1.EventEnvelope) (domain.EventResult, error) { if err := s.requireRepository(); err != nil { return domain.EventResult{}, err @@ -160,8 +160,9 @@ func (s *Service) HandleRoomEvent(ctx context.Context, envelope *roomeventsv1.Ev if gift.GetIsRobotGift() { return domain.EventResult{EventID: envelope.GetEventId(), Status: domain.EventStatusSkipped}, nil } - if strings.TrimSpace(envelope.GetRoomId()) == "" || gift.GetCoinSpent() <= 0 { - // 缺房间或 0 金币不会形成有效房间贡献；这类事件本身没有可补偿数据，跳过比失败重放更安全。 + contributionValue := gift.GetGiftValue() + if strings.TrimSpace(envelope.GetRoomId()) == "" || contributionValue <= 0 { + // 房间流水奖励只消费 wallet 结算后的 HeatValue，也就是 RoomGiftSent.gift_value；原始扣费 coin_spent 不再参与达标口径。 return domain.EventResult{EventID: envelope.GetEventId(), Status: domain.EventStatusSkipped}, nil } occurred := time.UnixMilli(envelope.GetOccurredAtMs()).UTC() @@ -177,7 +178,7 @@ func (s *Service) HandleRoomEvent(ctx context.Context, envelope *roomeventsv1.Ev EventType: envelope.GetEventType(), AppCode: envelope.GetAppCode(), RoomID: strings.TrimSpace(envelope.GetRoomId()), - CoinSpent: gift.GetCoinSpent(), + CoinSpent: contributionValue, OccurredAtMS: occurred.UnixMilli(), }, start.UnixMilli(), end.UnixMilli(), s.now().UTC().UnixMilli()) } @@ -200,7 +201,7 @@ func (s *Service) ListSettlements(ctx context.Context, query domain.SettlementQu return s.repository.ListRoomTurnoverRewardSettlements(ctx, query) } -// ProcessSettlementBatch 先为上一 UTC 周期补齐 settlement，再处理 pending 发币。 +// ProcessSettlementBatch 先为上一完整贡献周期补齐 settlement，再处理 pending 发币。 func (s *Service) ProcessSettlementBatch(ctx context.Context, batchSize int) (domain.BatchResult, error) { if err := s.requireRepository(); err != nil { return domain.BatchResult{}, err @@ -216,7 +217,7 @@ func (s *Service) ProcessSettlementBatch(ctx context.Context, batchSize int) (do // 活动未配置或关闭时不创建 settlement，也不触碰历史 pending；运营重新开启后再由 cron 正常补偿。 return domain.BatchResult{}, nil } - // 只结算上一个完整 UTC 周期；当前周仍在持续接收 RoomGiftSent，不能提前冻结或清空。 + // 只结算上一个完整贡献周期；当前周期仍在持续接收 RoomGiftSent，不能提前冻结或清空。 periodStart, periodEnd := PreviousWeekWindow(s.now().UTC()) result := domain.BatchResult{} progresses, err := s.repository.ListUnsettledRoomTurnoverRewardProgress(ctx, periodStart.UnixMilli(), batchSize) @@ -500,15 +501,19 @@ func MatchHighestTier(tiers []domain.Tier, coinSpent int64) (domain.Tier, bool) return matched, found } -// WeekWindow 返回给定时间所在 UTC 周一 00:00 到下一周一 00:00 的左闭右开窗口。 +// WeekWindow 返回给定时间所在 UTC 周一 01:00 到下一周一 01:00 的左闭右开窗口。 func WeekWindow(now time.Time) (time.Time, time.Time) { utc := now.UTC() dayStart := time.Date(utc.Year(), utc.Month(), utc.Day(), 0, 0, 0, 0, time.UTC) + // 贡献周期比 UTC 0 点延后一小时切换，确保 00:00 后的结算任务仍能稳定处理上一周期榜单。 + if utc.Before(dayStart.Add(time.Hour)) { + dayStart = dayStart.AddDate(0, 0, -1) + } weekday := int(dayStart.Weekday()) if weekday == 0 { weekday = 7 } - start := dayStart.AddDate(0, 0, 1-weekday) + start := dayStart.AddDate(0, 0, 1-weekday).Add(time.Hour) return start, start.AddDate(0, 0, 7) } diff --git a/services/activity-service/internal/service/roomturnoverreward/service_test.go b/services/activity-service/internal/service/roomturnoverreward/service_test.go index 2858fe66..65b78fcb 100644 --- a/services/activity-service/internal/service/roomturnoverreward/service_test.go +++ b/services/activity-service/internal/service/roomturnoverreward/service_test.go @@ -6,12 +6,15 @@ import ( "time" "google.golang.org/grpc" + "google.golang.org/protobuf/proto" + roomeventsv1 "hyapp.local/api/proto/events/room/v1" roomv1 "hyapp.local/api/proto/room/v1" walletv1 "hyapp.local/api/proto/wallet/v1" + "hyapp/pkg/appcode" domain "hyapp/services/activity-service/internal/domain/roomturnoverreward" ) -func TestWeekWindowUsesUTCMondayBoundary(t *testing.T) { +func TestWeekWindowUsesUTCOneMondayBoundary(t *testing.T) { cases := []struct { name string now string @@ -19,28 +22,34 @@ func TestWeekWindowUsesUTCMondayBoundary(t *testing.T) { wantEnd string }{ { - name: "monday zero belongs to new week", + name: "monday zero still belongs to previous week", now: "2026-06-01T00:00:00Z", - wantStart: "2026-06-01T00:00:00Z", - wantEnd: "2026-06-08T00:00:00Z", + wantStart: "2026-05-25T01:00:00Z", + wantEnd: "2026-06-01T01:00:00Z", + }, + { + name: "monday one starts new contribution week", + now: "2026-06-01T01:00:00Z", + wantStart: "2026-06-01T01:00:00Z", + wantEnd: "2026-06-08T01:00:00Z", }, { name: "sunday belongs to current week", now: "2026-06-07T23:59:59Z", - wantStart: "2026-06-01T00:00:00Z", - wantEnd: "2026-06-08T00:00:00Z", + wantStart: "2026-06-01T01:00:00Z", + wantEnd: "2026-06-08T01:00:00Z", }, { name: "cross month keeps utc week", now: "2026-07-01T12:30:00Z", - wantStart: "2026-06-29T00:00:00Z", - wantEnd: "2026-07-06T00:00:00Z", + wantStart: "2026-06-29T01:00:00Z", + wantEnd: "2026-07-06T01:00:00Z", }, { name: "cross year keeps utc week", now: "2027-01-01T08:00:00Z", - wantStart: "2026-12-28T00:00:00Z", - wantEnd: "2027-01-04T00:00:00Z", + wantStart: "2026-12-28T01:00:00Z", + wantEnd: "2027-01-04T01:00:00Z", }, } @@ -59,11 +68,11 @@ func TestWeekWindowUsesUTCMondayBoundary(t *testing.T) { } func TestPreviousWeekWindowReturnsCompletedUTCWeek(t *testing.T) { - start, end := PreviousWeekWindow(mustParseTime(t, "2026-06-08T00:00:00Z")) - if got := start.Format(time.RFC3339); got != "2026-06-01T00:00:00Z" { + start, end := PreviousWeekWindow(mustParseTime(t, "2026-06-08T01:00:00Z")) + if got := start.Format(time.RFC3339); got != "2026-06-01T01:00:00Z" { t.Fatalf("unexpected previous start: %s", got) } - if got := end.Format(time.RFC3339); got != "2026-06-08T00:00:00Z" { + if got := end.Format(time.RFC3339); got != "2026-06-08T01:00:00Z" { t.Fatalf("unexpected previous end: %s", got) } } @@ -118,6 +127,39 @@ func TestRetrySettlementRejectsTierNotMatchedWithoutGrantSideEffects(t *testing. } } +func TestHandleRoomEventConsumesGiftValueAsHeatValue(t *testing.T) { + repo := &roomGiftEventRepository{} + svc := New(repo, nil, nil) + svc.SetClock(func() time.Time { return mustParseTime(t, "2026-06-03T12:00:00Z") }) + body, err := proto.Marshal(&roomeventsv1.RoomGiftSent{ + GiftValue: 900, + CoinSpent: 1200, + }) + if err != nil { + t.Fatalf("marshal gift: %v", err) + } + result, err := svc.HandleRoomEvent(context.Background(), &roomeventsv1.EventEnvelope{ + EventId: "room-gift-heat-value", + EventType: "RoomGiftSent", + RoomId: "room-1", + AppCode: appcode.Default, + OccurredAtMs: mustParseTime(t, "2026-06-03T10:00:00Z").UnixMilli(), + Body: body, + }) + if err != nil { + t.Fatalf("HandleRoomEvent returned error: %v", err) + } + if result.Status != domain.EventStatusConsumed { + t.Fatalf("unexpected result: %+v", result) + } + if repo.event.CoinSpent != 900 { + t.Fatalf("room turnover reward must consume gift_value/HeatValue, got %+v", repo.event) + } + if got := time.UnixMilli(repo.periodStartMS).UTC().Format(time.RFC3339); got != "2026-06-01T01:00:00Z" { + t.Fatalf("unexpected period start: %s", got) + } +} + func mustParseTime(t *testing.T, value string) time.Time { t.Helper() parsed, err := time.Parse(time.RFC3339, value) @@ -134,6 +176,67 @@ type retryBoundaryRepository struct { failCalls int } +type roomGiftEventRepository struct { + event domain.RoomGiftEvent + periodStartMS int64 + periodEndMS int64 +} + +func (r *roomGiftEventRepository) GetRoomTurnoverRewardConfig(context.Context) (domain.Config, bool, error) { + return domain.Config{}, false, nil +} + +func (r *roomGiftEventRepository) UpdateRoomTurnoverRewardConfig(context.Context, domain.Config, int64) (domain.Config, error) { + return domain.Config{}, nil +} + +func (r *roomGiftEventRepository) ConsumeRoomTurnoverGiftEvent(_ context.Context, event domain.RoomGiftEvent, periodStartMS int64, periodEndMS int64, _ int64) (domain.EventResult, error) { + r.event = event + r.periodStartMS = periodStartMS + r.periodEndMS = periodEndMS + return domain.EventResult{EventID: event.EventID, Status: domain.EventStatusConsumed}, nil +} + +func (r *roomGiftEventRepository) GetRoomTurnoverRewardProgress(context.Context, string, int64) (domain.Progress, bool, error) { + return domain.Progress{}, false, nil +} + +func (r *roomGiftEventRepository) GetLatestRoomTurnoverRewardSettlement(context.Context, string) (domain.Settlement, bool, error) { + return domain.Settlement{}, false, nil +} + +func (r *roomGiftEventRepository) ListRoomTurnoverRewardSettlements(context.Context, domain.SettlementQuery) ([]domain.Settlement, int64, error) { + return nil, 0, nil +} + +func (r *roomGiftEventRepository) ListUnsettledRoomTurnoverRewardProgress(context.Context, int64, int) ([]domain.Progress, error) { + return nil, nil +} + +func (r *roomGiftEventRepository) InsertRoomTurnoverRewardSettlement(context.Context, domain.Settlement, int64) (domain.Settlement, bool, error) { + return domain.Settlement{}, false, nil +} + +func (r *roomGiftEventRepository) ListPendingRoomTurnoverRewardSettlements(context.Context, int) ([]domain.Settlement, error) { + return nil, nil +} + +func (r *roomGiftEventRepository) GetRoomTurnoverRewardSettlement(context.Context, string) (domain.Settlement, bool, error) { + return domain.Settlement{}, false, nil +} + +func (r *roomGiftEventRepository) ResetRoomTurnoverRewardSettlementForRetry(context.Context, string, int64, int64) (domain.Settlement, error) { + return domain.Settlement{}, nil +} + +func (r *roomGiftEventRepository) MarkRoomTurnoverRewardSettlementGranted(context.Context, string, string, int64) (domain.Settlement, error) { + return domain.Settlement{}, nil +} + +func (r *roomGiftEventRepository) MarkRoomTurnoverRewardSettlementFailed(context.Context, string, string, int64) error { + return nil +} + func (r *retryBoundaryRepository) GetRoomTurnoverRewardConfig(context.Context) (domain.Config, bool, error) { panic("unexpected GetRoomTurnoverRewardConfig") } diff --git a/services/activity-service/internal/storage/mysql/agency_opening_migration.go b/services/activity-service/internal/storage/mysql/agency_opening_migration.go index 06045642..db0d5a62 100644 --- a/services/activity-service/internal/storage/mysql/agency_opening_migration.go +++ b/services/activity-service/internal/storage/mysql/agency_opening_migration.go @@ -29,7 +29,7 @@ func (r *Repository) ensureAgencyOpeningTables(ctx context.Context) error { app_code VARCHAR(32) NOT NULL DEFAULT 'lalu' COMMENT '应用编码', cycle_id VARCHAR(96) NOT NULL COMMENT '代理开业周期 ID', rank_no INT NOT NULL COMMENT '档位序号', - threshold_coin_spent BIGINT NOT NULL DEFAULT 0 COMMENT '命中该档需要的代理房间流水金币', + threshold_coin_spent BIGINT NOT NULL DEFAULT 0 COMMENT '命中该档需要的代理房间贡献值 HeatValue', reward_coin_amount BIGINT 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', @@ -45,7 +45,7 @@ func (r *Repository) ensureAgencyOpeningTables(ctx context.Context) error { host_count INT NOT NULL DEFAULT 0 COMMENT '申请时有效主播数快照', agency_created_at_ms BIGINT NOT NULL DEFAULT 0 COMMENT '代理创建时间快照', status VARCHAR(32) NOT NULL DEFAULT 'applied' COMMENT 'applied/pending/granted/failed', - score_coin_amount BIGINT NOT NULL DEFAULT 0 COMMENT '周期内开业流水金币', + score_coin_amount BIGINT NOT NULL DEFAULT 0 COMMENT '历史字段名：周期内开业房间贡献值 HeatValue', reward_rank_no INT NOT NULL DEFAULT 0 COMMENT '结算命中的档位序号', reward_threshold_coin_spent BIGINT NOT NULL DEFAULT 0 COMMENT '结算命中的流水阈值', reward_coin_amount BIGINT NOT NULL DEFAULT 0 COMMENT '结算奖励金币', @@ -71,7 +71,7 @@ func (r *Repository) ensureAgencyOpeningTables(ctx context.Context) error { application_id VARCHAR(96) NOT NULL COMMENT '命中的申请 ID', agency_id BIGINT NOT NULL COMMENT '代理 ID', target_user_id BIGINT NOT NULL COMMENT '收礼主播用户 ID', - score_delta BIGINT NOT NULL COMMENT '本次流水金币', + score_delta BIGINT NOT NULL COMMENT '本次房间贡献值 HeatValue', occurred_at_ms BIGINT NOT NULL COMMENT '事件发生时间，UTC epoch ms', created_at_ms BIGINT NOT NULL COMMENT '创建时间，UTC epoch ms', PRIMARY KEY (app_code, source_event_id), diff --git a/services/activity-service/internal/storage/mysql/room_turnover_reward_migration.go b/services/activity-service/internal/storage/mysql/room_turnover_reward_migration.go index bbebcdee..8bbe9bf1 100644 --- a/services/activity-service/internal/storage/mysql/room_turnover_reward_migration.go +++ b/services/activity-service/internal/storage/mysql/room_turnover_reward_migration.go @@ -17,7 +17,7 @@ func (r *Repository) ensureRoomTurnoverRewardTables(ctx context.Context) error { app_code VARCHAR(32) NOT NULL DEFAULT 'lalu' COMMENT '应用编码', tier_code VARCHAR(64) NOT NULL COMMENT '档位编码', tier_name VARCHAR(128) NOT NULL COMMENT '档位名称', - threshold_coin_spent BIGINT NOT NULL COMMENT '命中该档需要的房间送礼金币流水', + threshold_coin_spent BIGINT NOT NULL COMMENT '命中该档需要的房间贡献值 HeatValue', reward_coin_amount BIGINT NOT NULL COMMENT '奖励房主金币数', status VARCHAR(32) NOT NULL DEFAULT 'active' COMMENT 'active/inactive', sort_order INT NOT NULL DEFAULT 0 COMMENT '排序', @@ -33,7 +33,7 @@ func (r *Repository) ensureRoomTurnoverRewardTables(ctx context.Context) error { event_type VARCHAR(96) NOT NULL COMMENT '事件类型', room_id VARCHAR(96) NOT NULL COMMENT '房间 ID', period_start_ms BIGINT NOT NULL COMMENT 'UTC 周期开始', - coin_spent BIGINT NOT NULL COMMENT '本事件送礼金币消耗', + coin_spent BIGINT NOT NULL COMMENT '历史字段名：本事件房间贡献值 HeatValue', status VARCHAR(32) NOT NULL COMMENT '消费状态', consumed_at_ms BIGINT NOT NULL COMMENT '消费时间，UTC epoch ms', created_at_ms BIGINT NOT NULL COMMENT '创建时间，UTC epoch ms', @@ -46,7 +46,7 @@ func (r *Repository) ensureRoomTurnoverRewardTables(ctx context.Context) error { room_id VARCHAR(96) NOT NULL COMMENT '房间 ID', period_start_ms BIGINT NOT NULL COMMENT 'UTC 周期开始', period_end_ms BIGINT NOT NULL COMMENT 'UTC 周期结束', - coin_spent BIGINT NOT NULL DEFAULT 0 COMMENT '本周期房间送礼金币流水', + coin_spent BIGINT NOT NULL DEFAULT 0 COMMENT '历史字段名：本周期房间贡献值 HeatValue', last_event_at_ms BIGINT 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', @@ -60,7 +60,7 @@ func (r *Repository) ensureRoomTurnoverRewardTables(ctx context.Context) error { owner_user_id BIGINT NOT NULL DEFAULT 0 COMMENT '结算收款房主', period_start_ms BIGINT NOT NULL COMMENT 'UTC 周期开始', period_end_ms BIGINT NOT NULL COMMENT 'UTC 周期结束', - coin_spent BIGINT NOT NULL DEFAULT 0 COMMENT '周期房间流水', + coin_spent BIGINT NOT NULL DEFAULT 0 COMMENT '历史字段名：周期房间贡献值 HeatValue', tier_id BIGINT NOT NULL DEFAULT 0 COMMENT '命中档位 ID', tier_code VARCHAR(64) NOT NULL DEFAULT '' COMMENT '命中档位编码', threshold_coin_spent BIGINT NOT NULL DEFAULT 0 COMMENT '命中档位阈值', diff --git a/services/room-service/deploy/mysql/initdb/001_room_service.sql b/services/room-service/deploy/mysql/initdb/001_room_service.sql index 4162b15f..807db00a 100644 --- a/services/room-service/deploy/mysql/initdb/001_room_service.sql +++ b/services/room-service/deploy/mysql/initdb/001_room_service.sql @@ -47,6 +47,8 @@ CREATE TABLE IF NOT EXISTS room_list_entries ( closed_by_admin_name VARCHAR(128) NOT NULL DEFAULT '' COMMENT '关闭操作管理员名称', closed_at_ms BIGINT NOT NULL DEFAULT 0 COMMENT '关闭时间，UTC epoch ms', heat BIGINT NOT NULL DEFAULT 0 COMMENT '热度值', + weekly_contribution BIGINT NOT NULL DEFAULT 0 COMMENT '当前贡献周期房间贡献，周一 01:00 UTC 自动切新周期', + contribution_week_start_ms BIGINT NOT NULL DEFAULT 0 COMMENT 'weekly_contribution 所属 UTC 周一 01:00 毫秒', online_count INT NOT NULL DEFAULT 0 COMMENT '在线数量', seat_count INT NOT NULL DEFAULT 0 COMMENT '麦位数量', occupied_seat_count INT NOT NULL DEFAULT 0 COMMENT '占用麦位数量', @@ -59,6 +61,7 @@ CREATE TABLE IF NOT EXISTS room_list_entries ( KEY idx_room_list_region_new (app_code, visible_region_id, status, created_at_ms DESC, room_id), KEY idx_room_list_region_country_hot (app_code, visible_region_id, owner_country_code, status, sort_score DESC, room_id), KEY idx_room_list_region_country_new (app_code, visible_region_id, owner_country_code, status, created_at_ms DESC, room_id), + KEY idx_room_list_weekly_contribution (app_code, status, visible_region_id, weekly_contribution DESC, created_at_ms DESC, room_id), KEY idx_room_list_owner (app_code, owner_user_id, updated_at_ms) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='房间列表项表'; diff --git a/services/room-service/internal/room/service/admin.go b/services/room-service/internal/room/service/admin.go index 2b262845..2d7bcd6f 100644 --- a/services/room-service/internal/room/service/admin.go +++ b/services/room-service/internal/room/service/admin.go @@ -29,6 +29,7 @@ func (s *Service) AdminListRooms(ctx context.Context, req *roomv1.AdminListRooms OwnerUserID: req.GetOwnerUserId(), SortBy: strings.TrimSpace(req.GetSortBy()), SortDirection: strings.TrimSpace(req.GetSortDirection()), + NowMS: s.clock.Now().UnixMilli(), }) if err != nil { return nil, err @@ -42,7 +43,7 @@ func (s *Service) AdminListRooms(ctx context.Context, req *roomv1.AdminListRooms func (s *Service) AdminGetRoom(ctx context.Context, req *roomv1.AdminGetRoomRequest) (*roomv1.AdminGetRoomResponse, error) { ctx = appcode.WithContext(ctx, req.GetMeta().GetAppCode()) - room, exists, err := s.repository.AdminGetRoom(ctx, req.GetRoomId()) + room, exists, err := s.repository.AdminGetRoom(ctx, req.GetRoomId(), s.clock.Now().UnixMilli()) if err != nil { return nil, err } diff --git a/services/room-service/internal/room/service/gift.go b/services/room-service/internal/room/service/gift.go index dfd322aa..ce8ff85e 100644 --- a/services/room-service/internal/room/service/gift.go +++ b/services/room-service/internal/room/service/gift.go @@ -354,6 +354,12 @@ func (s *Service) sendGift(ctx context.Context, req *roomv1.SendGiftRequest, rob LastGiftAtMS: now.UnixMilli(), }, }, + roomWeeklyContribution: &RoomWeeklyContributionIncrement{ + AppCode: appcode.FromContext(ctx), + RoomID: current.RoomID, + GiftValue: heatValue, + OccurredMS: now.UnixMilli(), + }, syncEvent: &tencentim.RoomEvent{ // 同步广播只选 GiftSent 主事件作为客户端房间系统消息，Heat/Rank 通过字段携带。 EventID: giftEvents[0].EventID, @@ -390,6 +396,7 @@ func (s *Service) sendGift(ctx context.Context, req *roomv1.SendGiftRequest, rob // 机器人房间可以增加房间热度和麦位收礼热度，但不进入跨房礼物榜和房间送礼统计。 result.roomGiftLeaderboard = nil result.roomUserGiftStats = nil + result.roomWeeklyContribution = nil } return result, records, nil }) diff --git a/services/room-service/internal/room/service/gift_leaderboard_test.go b/services/room-service/internal/room/service/gift_leaderboard_test.go index 03bee376..ef458ff0 100644 --- a/services/room-service/internal/room/service/gift_leaderboard_test.go +++ b/services/room-service/internal/room/service/gift_leaderboard_test.go @@ -79,3 +79,133 @@ func TestSendGiftWritesRoomGiftLeaderboardWithHeatValue(t *testing.T) { t.Fatalf("leaderboard must use heat value after global ratio, got %+v", leaderboard.increments[0]) } } + +func TestAdminRoomContributionUsesCurrentUTCWeek(t *testing.T) { + ctx := context.Background() + repository := mysqltest.NewRepository(t) + wallet := &rocketTestWallet{debits: []*walletv1.DebitGiftResponse{ + { + BillingReceiptId: "receipt-week-one", + CoinSpent: 100, + ChargeAmount: 100, + HeatValue: 10, + GiftTypeCode: "normal", + }, + { + BillingReceiptId: "receipt-week-two", + CoinSpent: 70, + ChargeAmount: 70, + HeatValue: 7, + GiftTypeCode: "normal", + }, + }} + now := &fixedRoomRocketClock{now: time.Date(2026, 6, 5, 12, 0, 0, 0, time.UTC)} + svc := roomservice.New(roomservice.Config{ + NodeID: "node-admin-weekly-contribution-test", + LeaseTTL: 10 * time.Second, + RankLimit: 20, + SnapshotEveryN: 1, + Clock: now, + }, router.NewMemoryDirectory(), repository, wallet, integration.NewNoopRoomEventPublisher(), integration.NewNoopOutboxPublisher()) + + roomID := "room-admin-weekly-contribution" + senderID := int64(18101) + targetID := int64(18102) + createRocketRoom(t, ctx, svc, roomID, senderID, 9001) + joinRocketRoom(t, ctx, svc, roomID, targetID) + sendGiftForWeeklyContribution(t, ctx, svc, roomID, senderID, targetID, "cmd-week-one") + + firstWeek, err := svc.AdminListRooms(ctx, &roomv1.AdminListRoomsRequest{ + Meta: &roomv1.RequestMeta{AppCode: appcode.Default}, + Page: 1, + PageSize: 20, + SortBy: "room_contribution", + SortDirection: "desc", + }) + if err != nil { + t.Fatalf("list first week rooms failed: %v", err) + } + if got := adminRoomHeatByID(firstWeek.GetRooms(), roomID); got != 10 { + t.Fatalf("first week contribution mismatch: got %d want 10", got) + } + + now.now = time.Date(2026, 6, 8, 0, 0, 0, 0, time.UTC) + cutoverBuffer, err := svc.AdminListRooms(ctx, &roomv1.AdminListRoomsRequest{ + Meta: &roomv1.RequestMeta{AppCode: appcode.Default}, + Page: 1, + PageSize: 20, + SortBy: "room_contribution", + SortDirection: "desc", + }) + if err != nil { + t.Fatalf("list cutover buffer rooms failed: %v", err) + } + if got := adminRoomHeatByID(cutoverBuffer.GetRooms(), roomID); got != 10 { + t.Fatalf("monday 00:00 UTC must keep previous contribution until UTC+1 cutover: got %d want 10", got) + } + + now.now = time.Date(2026, 6, 8, 1, 0, 0, 0, time.UTC) + nextWeekBeforeGift, err := svc.AdminListRooms(ctx, &roomv1.AdminListRoomsRequest{ + Meta: &roomv1.RequestMeta{AppCode: appcode.Default}, + Page: 1, + PageSize: 20, + SortBy: "room_contribution", + SortDirection: "desc", + }) + if err != nil { + t.Fatalf("list next contribution week rooms failed: %v", err) + } + if got := adminRoomHeatByID(nextWeekBeforeGift.GetRooms(), roomID); got != 0 { + t.Fatalf("next UTC+1 week must clear contribution before new gifts: got %d want 0", got) + } + + resp := sendGiftForWeeklyContribution(t, ctx, svc, roomID, senderID, targetID, "cmd-week-two") + if resp.GetRoomHeat() != 17 { + t.Fatalf("room heat must remain cumulative Room Cell state: got %d want 17", resp.GetRoomHeat()) + } + nextWeekAfterGift, err := svc.AdminListRooms(ctx, &roomv1.AdminListRoomsRequest{ + Meta: &roomv1.RequestMeta{AppCode: appcode.Default}, + Page: 1, + PageSize: 20, + SortBy: "room_contribution", + SortDirection: "desc", + }) + if err != nil { + t.Fatalf("list next week rooms after gift failed: %v", err) + } + if got := adminRoomHeatByID(nextWeekAfterGift.GetRooms(), roomID); got != 7 { + t.Fatalf("next UTC+1 week contribution must restart from new gifts: got %d want 7", got) + } +} + +func sendGiftForWeeklyContribution(t *testing.T, ctx context.Context, svc *roomservice.Service, roomID string, senderID int64, targetID int64, commandID string) *roomv1.SendGiftResponse { + t.Helper() + + resp, err := svc.SendGift(ctx, &roomv1.SendGiftRequest{ + Meta: &roomv1.RequestMeta{ + RequestId: "req-" + commandID, + CommandId: commandID, + ActorUserId: senderID, + RoomId: roomID, + AppCode: appcode.Default, + SentAtMs: time.Now().UTC().UnixMilli(), + }, + TargetType: "user", + TargetUserId: targetID, + GiftId: "gift-weekly-contribution", + GiftCount: 1, + }) + if err != nil { + t.Fatalf("send gift %s failed: %v", commandID, err) + } + return resp +} + +func adminRoomHeatByID(items []*roomv1.AdminRoomListItem, roomID string) int64 { + for _, item := range items { + if item.GetRoomId() == roomID { + return item.GetHeat() + } + } + return -1 +} diff --git a/services/room-service/internal/room/service/pipeline.go b/services/room-service/internal/room/service/pipeline.go index aa7b6e4f..b5fb9e99 100644 --- a/services/room-service/internal/room/service/pipeline.go +++ b/services/room-service/internal/room/service/pipeline.go @@ -91,14 +91,15 @@ func (s *Service) mutateRoom(ctx context.Context, cmd command.Command, replayabl LeaseToken: lease.LeaseToken, CreatedAtMS: now.UnixMilli(), }, - OutboxRecords: outboxRecords, - RoomStatus: result.roomStatus, - RoomSeatCount: result.roomSeatCount, - RoomPasswordHash: result.roomPasswordHash, - RoomMode: result.roomMode, - VisibleRegionID: result.visibleRegionID, - CloseInfo: result.closeInfo, - RoomUserGiftStats: result.roomUserGiftStats, + OutboxRecords: outboxRecords, + RoomStatus: result.roomStatus, + RoomSeatCount: result.roomSeatCount, + RoomPasswordHash: result.roomPasswordHash, + RoomMode: result.roomMode, + VisibleRegionID: result.visibleRegionID, + CloseInfo: result.closeInfo, + RoomUserGiftStats: result.roomUserGiftStats, + RoomWeeklyContribution: result.roomWeeklyContribution, }); err != nil { return nil, err } diff --git a/services/room-service/internal/room/service/repository.go b/services/room-service/internal/room/service/repository.go index b335228a..8fbeb6a1 100644 --- a/services/room-service/internal/room/service/repository.go +++ b/services/room-service/internal/room/service/repository.go @@ -73,6 +73,8 @@ type MutationCommit struct { CloseInfo *RoomCloseInfo // RoomUserGiftStats 是本次命令产生的房间用户送礼价值增量，必须和命令日志同事务提交。 RoomUserGiftStats []RoomUserGiftStatIncrement + // RoomWeeklyContribution 是后台房间贡献列的 UTC 周期读模型增量；它和命令日志同事务提交，避免送礼成功后榜单缺分。 + RoomWeeklyContribution *RoomWeeklyContributionIncrement } // RoomUserGiftStatIncrement 表达当前房间内某个送礼用户的礼物价值增量。 @@ -84,6 +86,14 @@ type RoomUserGiftStatIncrement struct { LastGiftAtMS int64 } +// RoomWeeklyContributionIncrement 表达一个房间在当前 UTC 周内新增的展示贡献。 +type RoomWeeklyContributionIncrement struct { + AppCode string + RoomID string + GiftValue int64 + OccurredMS int64 +} + type RoomCloseInfo struct { Reason string AdminID uint64 @@ -200,6 +210,8 @@ type AdminRoomListQuery struct { OwnerUserID int64 SortBy string SortDirection string + // NowMS 用于把后台房间贡献收敛到当前 UTC 周；调用方传服务时钟，测试可以固定边界。 + NowMS int64 } type AdminRoomPinEntry struct { @@ -615,7 +627,7 @@ type Repository interface { // ListRoomOnlineUsers 查询房间 active 在线用户分页，不读取完整 RoomSnapshot。 ListRoomOnlineUsers(ctx context.Context, query RoomOnlineUserQuery) (RoomOnlineUserPage, error) AdminListRooms(ctx context.Context, query AdminRoomListQuery) ([]AdminRoomListEntry, int64, error) - AdminGetRoom(ctx context.Context, roomID string) (AdminRoomListEntry, bool, error) + AdminGetRoom(ctx context.Context, roomID string, nowMS int64) (AdminRoomListEntry, bool, error) AdminListRoomPins(ctx context.Context, query AdminRoomPinQuery) ([]AdminRoomPinEntry, int64, error) AdminCreateRoomPin(ctx context.Context, input AdminCreateRoomPinInput) (AdminRoomPinEntry, bool, error) AdminCancelRoomPin(ctx context.Context, pinID int64, adminID uint64, nowMS int64) (AdminRoomPinEntry, bool, error) diff --git a/services/room-service/internal/room/service/room_gift_leaderboard.go b/services/room-service/internal/room/service/room_gift_leaderboard.go index 81de3d84..38c98487 100644 --- a/services/room-service/internal/room/service/room_gift_leaderboard.go +++ b/services/room-service/internal/room/service/room_gift_leaderboard.go @@ -253,11 +253,15 @@ func roomGiftLeaderboardWindow(period string, now time.Time) (time.Time, time.Ti dayStart := time.Date(now.UTC().Year(), now.UTC().Month(), now.UTC().Day(), 0, 0, 0, 0, time.UTC) switch period { case roomGiftLeaderboardPeriodWeek: + // 周房间贡献榜按 UTC 周一 01:00 切周，避免 00:00 结算任务消费上一周期时 Redis 周榜已经换 key。 + if now.UTC().Before(dayStart.Add(time.Hour)) { + dayStart = dayStart.AddDate(0, 0, -1) + } weekday := int(dayStart.Weekday()) if weekday == 0 { weekday = 7 } - start := dayStart.AddDate(0, 0, 1-weekday) + start := dayStart.AddDate(0, 0, 1-weekday).Add(time.Hour) return start, start.AddDate(0, 0, 7) case roomGiftLeaderboardPeriodMonth: start := time.Date(dayStart.Year(), dayStart.Month(), 1, 0, 0, 0, 0, time.UTC) diff --git a/services/room-service/internal/room/service/service.go b/services/room-service/internal/room/service/service.go index a449db0c..f11cf887 100644 --- a/services/room-service/internal/room/service/service.go +++ b/services/room-service/internal/room/service/service.go @@ -146,6 +146,8 @@ type mutationResult struct { roomGiftLeaderboard *RoomGiftLeaderboardIncrement // roomUserGiftStats 是当前房间用户送礼价值统计，和命令日志同事务提交。 roomUserGiftStats []RoomUserGiftStatIncrement + // roomWeeklyContribution 是后台房间贡献列的当前 UTC 周增量，和命令日志同事务提交。 + roomWeeklyContribution *RoomWeeklyContributionIncrement } // New 初始化 room-service 领域服务。 diff --git a/services/room-service/internal/storage/mysql/repository.go b/services/room-service/internal/storage/mysql/repository.go index 9ee136b5..3dec2f4b 100644 --- a/services/room-service/internal/storage/mysql/repository.go +++ b/services/room-service/internal/storage/mysql/repository.go @@ -110,6 +110,8 @@ func (r *Repository) Migrate(ctx context.Context) error { closed_by_admin_name VARCHAR(128) NOT NULL DEFAULT '', closed_at_ms BIGINT NOT NULL DEFAULT 0, heat BIGINT NOT NULL DEFAULT 0, + weekly_contribution BIGINT NOT NULL DEFAULT 0, + contribution_week_start_ms BIGINT NOT NULL DEFAULT 0, online_count INT NOT NULL DEFAULT 0, seat_count INT NOT NULL DEFAULT 0, occupied_seat_count INT NOT NULL DEFAULT 0, @@ -122,6 +124,7 @@ func (r *Repository) Migrate(ctx context.Context) error { KEY idx_room_list_region_new (app_code, visible_region_id, status, created_at_ms DESC, room_id), KEY idx_room_list_region_country_hot (app_code, visible_region_id, owner_country_code, status, sort_score DESC, room_id), KEY idx_room_list_region_country_new (app_code, visible_region_id, owner_country_code, status, created_at_ms DESC, room_id), + KEY idx_room_list_weekly_contribution (app_code, status, visible_region_id, weekly_contribution DESC, created_at_ms DESC, room_id), KEY idx_room_list_owner (app_code, owner_user_id, updated_at_ms) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, `CREATE TABLE IF NOT EXISTS room_user_presence ( @@ -371,6 +374,9 @@ func (r *Repository) Migrate(ctx context.Context) error { if err := r.ensureRoomAdminCloseSchema(ctx); err != nil { return err } + if err := r.ensureRoomWeeklyContributionSchema(ctx); err != nil { + return err + } if err := r.ensureRoomUserPresenceSchema(ctx); err != nil { return err } @@ -381,6 +387,44 @@ func (r *Repository) Migrate(ctx context.Context) error { return r.ensureOutboxRetrySchema(ctx) } +func (r *Repository) ensureRoomWeeklyContributionSchema(ctx context.Context) error { + columns := []struct { + column string + statement string + }{ + { + column: "weekly_contribution", + statement: `ALTER TABLE room_list_entries ADD COLUMN weekly_contribution BIGINT NOT NULL DEFAULT 0 AFTER heat`, + }, + { + column: "contribution_week_start_ms", + statement: `ALTER TABLE room_list_entries ADD COLUMN contribution_week_start_ms BIGINT NOT NULL DEFAULT 0 AFTER weekly_contribution`, + }, + } + for _, item := range columns { + hasColumn, err := r.columnExists(ctx, "room_list_entries", item.column) + if err != nil { + return err + } + if !hasColumn { + if _, err := r.db.ExecContext(ctx, item.statement); err != nil { + return err + } + } + } + hasIndex, err := r.indexExists(ctx, "room_list_entries", "idx_room_list_weekly_contribution") + if err != nil { + return err + } + if !hasIndex { + // 后台按房间贡献排序时只读当前 UTC 周贡献；索引用于收敛 active/region 过滤后的排序成本。 + if _, err := r.db.ExecContext(ctx, `ALTER TABLE room_list_entries ADD INDEX idx_room_list_weekly_contribution (app_code, status, visible_region_id, weekly_contribution DESC, created_at_ms DESC, room_id)`); err != nil { + return err + } + } + return nil +} + func (r *Repository) ensureRoomPinSchema(ctx context.Context) error { hasColumn, err := r.columnExists(ctx, "room_region_pins", "pin_type") if err != nil { @@ -995,6 +1039,38 @@ func (r *Repository) SaveMutation(ctx context.Context, commit roomservice.Mutati return err } } + if weekly := commit.RoomWeeklyContribution; weekly != nil && weekly.GiftValue > 0 { + weeklyAppCode := normalizedRecordAppCode(ctx, weekly.AppCode) + weeklyRoomID := strings.TrimSpace(weekly.RoomID) + if weeklyRoomID == "" { + weeklyRoomID = commit.Command.RoomID + } + occurredMS := weekly.OccurredMS + if occurredMS <= 0 { + occurredMS = commit.Command.CreatedAtMS + } + weekStartMS := roomContributionWeekStartMS(occurredMS) + if _, err := tx.ExecContext(ctx, + `UPDATE room_list_entries + SET weekly_contribution = CASE + WHEN contribution_week_start_ms = ? THEN weekly_contribution + ? + ELSE ? + END, + contribution_week_start_ms = ?, + updated_at_ms = ? + WHERE app_code = ? AND room_id = ?`, + weekStartMS, + weekly.GiftValue, + weekly.GiftValue, + weekStartMS, + commit.Command.CreatedAtMS, + weeklyAppCode, + weeklyRoomID, + ); err != nil { + _ = tx.Rollback() + return err + } + } return tx.Commit() } @@ -2351,12 +2427,16 @@ func (r *Repository) AdminListRooms(ctx context.Context, query roomservice.Admin if query.PageSize > 100 { query.PageSize = 100 } + if query.NowMS <= 0 { + query.NowMS = time.Now().UTC().UnixMilli() + } where, args := adminRoomWhere(ctx, query) var total int64 if err := r.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM room_list_entries rle `+where, args...).Scan(&total); err != nil { return nil, 0, err } - rows, err := r.db.QueryContext(ctx, adminRoomSelectSQL()+where+` ORDER BY `+adminRoomOrderBy(query)+` LIMIT ? OFFSET ?`, append(args, query.PageSize, (query.Page-1)*query.PageSize)...) + contributionExpr := adminRoomContributionExpr(query.NowMS) + rows, err := r.db.QueryContext(ctx, adminRoomSelectSQL(contributionExpr)+where+` ORDER BY `+adminRoomOrderBy(query, contributionExpr)+` LIMIT ? OFFSET ?`, append(args, query.PageSize, (query.Page-1)*query.PageSize)...) if err != nil { return nil, 0, err } @@ -2375,8 +2455,9 @@ func (r *Repository) AdminListRooms(ctx context.Context, query roomservice.Admin return items, total, nil } -func (r *Repository) AdminGetRoom(ctx context.Context, roomID string) (roomservice.AdminRoomListEntry, bool, error) { - row := r.db.QueryRowContext(ctx, adminRoomSelectSQL()+` WHERE rle.app_code = ? AND rle.room_id = ?`, appcode.FromContext(ctx), strings.TrimSpace(roomID)) +func (r *Repository) AdminGetRoom(ctx context.Context, roomID string, nowMS int64) (roomservice.AdminRoomListEntry, bool, error) { + contributionExpr := adminRoomContributionExpr(nowMS) + row := r.db.QueryRowContext(ctx, adminRoomSelectSQL(contributionExpr)+` WHERE rle.app_code = ? AND rle.room_id = ?`, appcode.FromContext(ctx), strings.TrimSpace(roomID)) item, err := scanAdminRoom(row) if errors.Is(err, sql.ErrNoRows) { return roomservice.AdminRoomListEntry{}, false, nil @@ -2387,12 +2468,12 @@ func (r *Repository) AdminGetRoom(ctx context.Context, roomID string) (roomservi return item, true, nil } -func adminRoomSelectSQL() string { +func adminRoomSelectSQL(contributionExpr string) string { return ` SELECT rle.room_id, rle.room_short_id, rle.visible_region_id, rle.owner_user_id, rle.title, rle.cover_url, rle.mode, rle.status, rle.close_reason, rle.closed_by_admin_id, rle.closed_by_admin_name, rle.closed_at_ms, - rle.heat, rle.online_count, rle.seat_count, rle.occupied_seat_count, + ` + contributionExpr + ` AS room_contribution, rle.online_count, rle.seat_count, rle.occupied_seat_count, rle.sort_score, rle.created_at_ms, rle.updated_at_ms FROM room_list_entries rle ` } @@ -2429,19 +2510,42 @@ func adminRoomWhere(ctx context.Context, query roomservice.AdminRoomListQuery) ( return " WHERE " + strings.Join(where, " AND "), args } -func adminRoomOrderBy(query roomservice.AdminRoomListQuery) string { +func adminRoomOrderBy(query roomservice.AdminRoomListQuery, contributionExpr string) string { direction := "DESC" if strings.EqualFold(strings.TrimSpace(query.SortDirection), "asc") { direction = "ASC" } switch strings.TrimSpace(query.SortBy) { case "room_contribution", "heat": - return "rle.heat " + direction + ", rle.created_at_ms DESC, rle.room_id DESC" + return contributionExpr + " " + direction + ", rle.created_at_ms DESC, rle.room_id DESC" default: return "rle.created_at_ms DESC, rle.room_id DESC" } } +func adminRoomContributionExpr(nowMS int64) string { + weekStartMS := roomContributionWeekStartMS(nowMS) + return "CASE WHEN rle.contribution_week_start_ms = " + strconv.FormatInt(weekStartMS, 10) + " THEN rle.weekly_contribution ELSE 0 END" +} + +func roomContributionWeekStartMS(nowMS int64) int64 { + if nowMS <= 0 { + nowMS = time.Now().UTC().UnixMilli() + } + dayStart := time.UnixMilli(nowMS).UTC() + dayStart = time.Date(dayStart.Year(), dayStart.Month(), dayStart.Day(), 0, 0, 0, 0, time.UTC) + // 房间贡献榜按 UTC 周一 01:00 切周，给 00:00 后的结算和榜单消费留出缓冲窗口。 + cutoff := dayStart.Add(time.Hour) + if time.UnixMilli(nowMS).UTC().Before(cutoff) { + dayStart = dayStart.AddDate(0, 0, -1) + } + weekday := int(dayStart.Weekday()) + if weekday == 0 { + weekday = 7 + } + return dayStart.AddDate(0, 0, 1-weekday).Add(time.Hour).UnixMilli() +} + func scanAdminRoom(scanner interface{ Scan(dest ...any) error }) (roomservice.AdminRoomListEntry, error) { var item roomservice.AdminRoomListEntry err := scanner.Scan( @@ -2501,7 +2605,7 @@ func (r *Repository) AdminListRoomPins(ctx context.Context, query roomservice.Ad func (r *Repository) AdminCreateRoomPin(ctx context.Context, input roomservice.AdminCreateRoomPinInput) (roomservice.AdminRoomPinEntry, bool, error) { app := appcode.FromContext(ctx) - room, exists, err := r.AdminGetRoom(ctx, input.RoomID) + room, exists, err := r.AdminGetRoom(ctx, input.RoomID, input.NowMS) if err != nil || !exists { return roomservice.AdminRoomPinEntry{}, false, err }