2026-06-03 18:19:50 +08:00

545 lines
21 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

package service
import (
"context"
"fmt"
"strconv"
"strings"
"time"
activityv1 "hyapp.local/api/proto/activity/v1"
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"
"hyapp/pkg/tencentim"
"hyapp/pkg/xerr"
"hyapp/services/room-service/internal/room/command"
"hyapp/services/room-service/internal/room/outbox"
"hyapp/services/room-service/internal/room/rank"
"hyapp/services/room-service/internal/room/state"
)
// SendGift 先同步调用 wallet再在 Room Cell 内完成热度、榜单和房间事件落地。
func (s *Service) SendGift(ctx context.Context, req *roomv1.SendGiftRequest) (*roomv1.SendGiftResponse, error) {
ctx = contextFromMeta(ctx, req.GetMeta())
// SendGift 的账务不在 room-service 内结算,但房间表现、热度和榜单由 Room Cell 同步更新。
cmd := command.SendGift{
Base: baseFromMeta(req.GetMeta()),
TargetType: normalizeGiftTargetType(req.GetTargetType()),
TargetUserIDs: normalizeGiftTargetUserIDs(req.GetTargetUserId(), req.GetTargetUserIds()),
GiftID: req.GetGiftId(),
PoolID: strings.TrimSpace(req.GetPoolId()),
GiftCount: req.GetGiftCount(),
SenderRegionID: req.GetSenderRegionId(),
TargetIsHost: req.GetTargetIsHost(),
// 工资区域由 gateway 从 user-service host profile 注入;房间 visible_region_id 只用于房间/礼物可见性。
TargetHostRegionID: req.GetTargetHostRegionId(),
TargetAgencyOwnerUserID: req.GetTargetAgencyOwnerUserId(),
}
if cmd.TargetType == "" {
cmd.TargetType = "user"
}
if cmd.TargetType == "user" && len(cmd.TargetUserIDs) > 0 {
cmd.TargetUserID = cmd.TargetUserIDs[0]
}
result, err := s.mutateRoom(ctx, cmd, true, func(now time.Time, current *state.RoomState, localRank *rank.LocalRank) (mutationResult, []outbox.Record, error) {
if _, exists := current.OnlineUsers[cmd.ActorUserID()]; !exists {
// 送礼者必须在房间业务 presence 内。
return mutationResult{}, nil, xerr.New(xerr.NotFound, "sender not in room")
}
if cmd.TargetType != "user" {
// v1 HTTP 契约已预留 all_mic/all_room/couple账务拆单和房间表现策略补齐前先 fail-close。
return mutationResult{}, nil, xerr.New(xerr.InvalidArgument, "target_type is unsupported")
}
if len(cmd.TargetUserIDs) == 0 || cmd.TargetUserID <= 0 {
return mutationResult{}, nil, xerr.New(xerr.InvalidArgument, "target_user_ids is required")
}
for _, targetUserID := range cmd.TargetUserIDs {
if _, exists := current.OnlineUsers[targetUserID]; !exists {
// 每个接收方都必须在房间 presence 内;批量扣费前先全部校验,避免钱包出现无房间表现的入账。
return mutationResult{}, nil, xerr.New(xerr.NotFound, "target not in room")
}
}
if cmd.GiftID == "" || cmd.GiftCount <= 0 {
// 礼物 ID 和数量是钱包查服务端价格的最小输入,客户端不再提交礼物单价。
return mutationResult{}, nil, xerr.New(xerr.InvalidArgument, "gift_id and gift_count must be valid")
}
if len(cmd.TargetUserIDs) > 1 && cmd.PoolID != "" {
// 幸运礼物 activity 协议当前只有单个 target_user_id多目标先只开放普通礼物避免抽奖事实错绑接收方。
return mutationResult{}, nil, xerr.New(xerr.InvalidArgument, "lucky gift requires one target user")
}
roomMeta, exists, err := s.repository.GetRoomMeta(ctx, cmd.RoomID())
if err != nil {
return mutationResult{}, nil, err
}
if !exists {
return mutationResult{}, nil, xerr.New(xerr.NotFound, "room not found")
}
luckyEnabled := false
if cmd.PoolID != "" {
// 客户端显式传 pool_id 时必须先检查活动规则;规则未启用就拒绝,避免先扣费再发现不能抽奖。
if s.luckyGift == nil {
return mutationResult{}, nil, xerr.New(xerr.Unavailable, "lucky gift service is not configured")
}
checkResp, err := s.luckyGift.CheckLuckyGift(ctx, &activityv1.CheckLuckyGiftRequest{
Meta: activityMetaFromRoom(ctx, req.GetMeta()),
UserId: cmd.ActorUserID(),
RoomId: cmd.RoomID(),
GiftId: cmd.GiftID,
PoolId: cmd.PoolID,
})
if err != nil {
return mutationResult{}, nil, err
}
if checkResp == nil || !checkResp.GetEnabled() {
return mutationResult{}, nil, xerr.New(xerr.RuleNotActive, "lucky gift rule is not active")
}
// Check 只说明“允许按该 pool 抽奖”,真实价格、区域可用性和扣费仍以 wallet-service 为准。
luckyEnabled = true
}
treasureConfig, err := s.roomTreasureConfig(ctx)
if err != nil {
return mutationResult{}, nil, err
}
var billing *walletv1.DebitGiftResponse
var targetBillings []giftTargetBilling
var walletDebitMS int64
settledCommand := cmd
var luckyGift *roomv1.LuckyGiftDrawResult
if err := s.withLuckyGiftSendLock(ctx, luckyEnabled, cmd.ActorUserID(), func() error {
// 钱包扣费在房间状态变更前完成;失败时不写 command log、不进入 Room Cell 提交态。
walletStartedAt := time.Now()
var err error
billing, targetBillings, err = s.debitGiftTargets(ctx, cmd, roomMeta)
walletDebitMS = elapsedMS(walletStartedAt)
if err != nil {
return err
}
if billing == nil {
return xerr.New(xerr.Unavailable, "wallet debit response is empty")
}
settledCommand.BillingReceiptID = billing.GetBillingReceiptId()
settledCommand.CoinSpent = billing.GetCoinSpent()
settledCommand.GiftPointAdded = billing.GetGiftPointAdded()
settledCommand.HeatValue = billing.GetHeatValue()
settledCommand.PriceVersion = billing.GetPriceVersion()
settledCommand.GiftTypeCode = billing.GetGiftTypeCode()
settledCommand.HostPeriodDiamondAdded = billing.GetHostPeriodDiamondAdded()
settledCommand.HostPeriodCycleKey = billing.GetHostPeriodCycleKey()
if !luckyEnabled && len(cmd.TargetUserIDs) == 1 {
// 未显式传 pool_id 的旧客户端仍可通过钱包礼物类型触发默认幸运礼物逻辑;新客户端应传明确 pool_id。
luckyEnabled = s.shouldDrawLuckyGift(cmd.PoolID, billing.GetGiftTypeCode())
}
if luckyEnabled {
// 抽奖必须发生在扣费成功之后activity-service 只接收钱包结算后的 coin_spent不信任客户端价格。
drawResp, err := s.luckyGift.ExecuteLuckyGiftDraw(ctx, &activityv1.ExecuteLuckyGiftDrawRequest{
LuckyGift: &activityv1.LuckyGiftMeta{
Meta: activityMetaFromRoom(ctx, req.GetMeta()),
CommandId: cmd.ID(),
UserId: cmd.ActorUserID(),
TargetUserId: cmd.TargetUserID,
// 目前没有独立设备 ID 字段,优先用 session_id没有时退回 command_id保证风控 scope 不为空。
DeviceId: luckyGiftDeviceID(cmd),
RoomId: cmd.RoomID(),
AnchorId: luckyGiftAnchorID(roomMeta),
GiftId: cmd.GiftID,
GiftCount: cmd.GiftCount,
CoinSpent: billing.GetCoinSpent(),
// 房间链路统一使用 UTC epoch ms不能把本地时区时间传给活动风控窗口。
PaidAtMs: now.UTC().UnixMilli(),
PoolId: cmd.PoolID,
VisibleRegionId: roomMeta.VisibleRegionID,
},
})
if err != nil {
// 抽奖失败时整笔送礼失败并依赖 wallet 幂等重试;不能落普通礼物后丢失幸运礼物抽奖事实。
return err
}
if drawResp == nil {
return xerr.New(xerr.Unavailable, "lucky gift draw response is empty")
}
// 同步返回只给当前送礼用户做即时表现;金币余额以 activity/wallet 快路径回执为准,房间开奖结果仍由 activity outbox 补偿。
luckyGift = luckyGiftResultFromProto(drawResp.GetResult())
}
return nil
}); err != nil {
return mutationResult{}, nil, err
}
heatValue := billing.GetHeatValue()
// 扣费成功后Room Cell 同步更新热度和本地礼物榜。
current.Heat += heatValue
current.GiftRank = localRank.ApplyGift(cmd.ActorUserID(), heatValue, now)
current.Version++
treasureApply, err := s.applyRoomTreasureGift(now, current, treasureConfig, cmd, &settledCommand, billing, roomMeta)
if err != nil {
return mutationResult{}, nil, err
}
commandPayload, err := command.Serialize(settledCommand)
if err != nil {
return mutationResult{}, nil, err
}
// 送礼事件按 target 拆分,保证 IM、统计和礼物墙消费者拿到准确的接收方和账务回执。
giftEvents := make([]outbox.Record, 0, len(targetBillings))
for _, targetBilling := range targetBillings {
giftEvent, err := outbox.Build(current.RoomID, "RoomGiftSent", current.Version, now, &roomeventsv1.RoomGiftSent{
SenderUserId: cmd.ActorUserID(),
TargetUserId: targetBilling.TargetUserID,
GiftId: cmd.GiftID,
PoolId: cmd.PoolID,
GiftCount: cmd.GiftCount,
GiftValue: targetBilling.Billing.GetHeatValue(),
BillingReceiptId: targetBilling.Billing.GetBillingReceiptId(),
VisibleRegionId: roomMeta.VisibleRegionID,
CommandId: targetBilling.CommandID,
})
if err != nil {
return mutationResult{}, nil, err
}
giftEvents = append(giftEvents, giftEvent)
}
heatEvent, err := outbox.Build(current.RoomID, "RoomHeatChanged", current.Version, now, &roomeventsv1.RoomHeatChanged{
Delta: heatValue,
CurrentHeat: current.Heat,
})
if err != nil {
return mutationResult{}, nil, err
}
rankItem := findRankItem(current.GiftRank, cmd.ActorUserID())
rankEvent, err := outbox.Build(current.RoomID, "RoomRankChanged", current.Version, now, &roomeventsv1.RoomRankChanged{
UserId: rankItem.UserID,
Score: rankItem.Score,
GiftValue: rankItem.GiftValue,
})
if err != nil {
return mutationResult{}, nil, err
}
records := make([]outbox.Record, 0, len(giftEvents)+4)
records = append(records, giftEvents...)
records = append(records, heatEvent, rankEvent)
if treasureApply.progressEvent != nil {
progressEvent, err := outbox.Build(current.RoomID, "RoomTreasureProgressChanged", current.Version, now, treasureApply.progressEvent)
if err != nil {
return mutationResult{}, nil, err
}
records = append(records, progressEvent)
}
if treasureApply.countdown != nil {
countdownEvent, err := outbox.Build(current.RoomID, "RoomTreasureCountdownStarted", current.Version, now, treasureApply.countdown)
if err != nil {
return mutationResult{}, nil, err
}
records = append(records, countdownEvent)
}
return mutationResult{
snapshot: current.ToProto(),
billingReceiptID: billing.GetBillingReceiptId(),
roomHeat: current.Heat,
giftRank: cloneProtoRank(current.GiftRank),
luckyGift: luckyGift,
commandPayload: commandPayload,
walletDebitMS: walletDebitMS,
roomGiftLeaderboard: &RoomGiftLeaderboardIncrement{
AppCode: appcode.FromContext(ctx),
RoomID: current.RoomID,
CoinSpent: roomGiftLeaderboardCoinSpent(billing),
OccurredAtMS: now.UnixMilli(),
},
roomUserGiftStats: []RoomUserGiftStatIncrement{
{
AppCode: appcode.FromContext(ctx),
RoomID: current.RoomID,
UserID: cmd.ActorUserID(),
GiftValue: heatValue,
LastGiftAtMS: now.UnixMilli(),
},
},
syncEvent: &tencentim.RoomEvent{
// 同步广播只选 GiftSent 主事件作为客户端房间系统消息Heat/Rank 通过字段携带。
EventID: giftEvents[0].EventID,
RoomID: current.RoomID,
EventType: "room_gift_sent",
ActorUserID: cmd.ActorUserID(),
TargetUserID: cmd.TargetUserID,
GiftValue: heatValue,
RoomHeat: current.Heat,
RoomVersion: current.Version,
Attributes: map[string]string{
"gift_id": cmd.GiftID,
"pool_id": cmd.PoolID,
"gift_count": fmt.Sprintf("%d", cmd.GiftCount),
"billing_receipt_id": billing.GetBillingReceiptId(),
"coin_spent": fmt.Sprintf("%d", settledCommand.CoinSpent),
"gift_point_added": fmt.Sprintf("%d", settledCommand.GiftPointAdded),
"target_count": fmt.Sprintf("%d", len(cmd.TargetUserIDs)),
"target_user_ids": giftTargetUserIDsAttribute(cmd.TargetUserIDs),
"price_version": settledCommand.PriceVersion,
"host_period_diamond_added": fmt.Sprintf("%d", settledCommand.HostPeriodDiamondAdded),
"host_period_cycle_key": settledCommand.HostPeriodCycleKey,
},
},
}, records, nil
})
if err != nil {
return nil, err
}
return &roomv1.SendGiftResponse{
Result: commandResult(result.applied, result.snapshot.GetVersion(), s.clock.Now()),
BillingReceiptId: result.billingReceiptID,
RoomHeat: result.roomHeat,
GiftRank: result.giftRank,
Room: result.snapshot,
Treasure: result.snapshot.GetTreasure(),
LuckyGift: result.luckyGift,
}, nil
}
type giftTargetBilling struct {
TargetUserID int64
CommandID string
Billing *walletv1.DebitGiftResponse
}
func (s *Service) debitGiftTargets(ctx context.Context, cmd command.SendGift, roomMeta RoomMeta) (*walletv1.DebitGiftResponse, []giftTargetBilling, error) {
if len(cmd.TargetUserIDs) == 1 {
// 单目标保留原 wallet command_id避免改变已有幂等键、账务回执和排障路径。
billing, err := s.wallet.DebitGift(ctx, &walletv1.DebitGiftRequest{
CommandId: cmd.ID(),
RoomId: cmd.RoomID(),
SenderUserId: cmd.ActorUserID(),
TargetUserId: cmd.TargetUserID,
GiftId: cmd.GiftID,
GiftCount: cmd.GiftCount,
AppCode: appcode.FromContext(ctx),
RegionId: roomMeta.VisibleRegionID,
SenderRegionId: cmd.SenderRegionID,
TargetIsHost: cmd.TargetIsHost,
TargetHostRegionId: cmd.TargetHostRegionID,
TargetAgencyOwnerUserId: cmd.TargetAgencyOwnerUserID,
})
if err != nil {
return nil, nil, err
}
return billing, []giftTargetBilling{{TargetUserID: cmd.TargetUserID, CommandID: cmd.ID(), Billing: billing}}, nil
}
targets := make([]*walletv1.DebitGiftTarget, 0, len(cmd.TargetUserIDs))
for _, targetUserID := range cmd.TargetUserIDs {
target := &walletv1.DebitGiftTarget{
CommandId: giftTargetCommandID(cmd.ID(), targetUserID, len(cmd.TargetUserIDs)),
TargetUserId: targetUserID,
}
if targetUserID == cmd.TargetUserID && cmd.TargetIsHost {
// 现有 room proto 只有单个 host scope多目标时只允许把该快照用于对应的第一个目标不能错误扩散到其他接收方。
target.TargetIsHost = true
target.TargetHostRegionId = cmd.TargetHostRegionID
target.TargetAgencyOwnerUserId = cmd.TargetAgencyOwnerUserID
}
targets = append(targets, target)
}
resp, err := s.wallet.BatchDebitGift(ctx, &walletv1.BatchDebitGiftRequest{
CommandId: cmd.ID(),
RoomId: cmd.RoomID(),
SenderUserId: cmd.ActorUserID(),
GiftId: cmd.GiftID,
GiftCount: cmd.GiftCount,
AppCode: appcode.FromContext(ctx),
RegionId: roomMeta.VisibleRegionID,
SenderRegionId: cmd.SenderRegionID,
Targets: targets,
})
if err != nil {
return nil, nil, err
}
targetBillings, err := giftTargetBillingsFromBatch(resp, cmd.TargetUserIDs)
if err != nil {
return nil, nil, err
}
return resp.GetAggregate(), targetBillings, nil
}
func roomGiftLeaderboardCoinSpent(billing *walletv1.DebitGiftResponse) int64 {
if billing == nil {
return 0
}
if billing.GetCoinSpent() > 0 {
return billing.GetCoinSpent()
}
return billing.GetChargeAmount()
}
func (s *Service) shouldDrawLuckyGift(poolID string, giftTypeCode string) bool {
if s.luckyGift == nil {
return false
}
switch strings.TrimSpace(giftTypeCode) {
case "lucky", "super_lucky":
return true
default:
return strings.TrimSpace(poolID) != ""
}
}
func (s *Service) withLuckyGiftSendLock(ctx context.Context, enabled bool, userID int64, fn func() error) error {
if !enabled || s == nil || s.luckyGiftSendLocker == nil {
return fn()
}
release, err := s.luckyGiftSendLocker.AcquireLuckyGiftSendLock(ctx, appcode.FromContext(ctx), userID, s.luckyGiftSendLockTTL)
if err != nil {
return err
}
if release != nil {
defer func() {
releaseCtx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
release(releaseCtx)
}()
}
return fn()
}
func activityMetaFromRoom(ctx context.Context, meta *roomv1.RequestMeta) *activityv1.RequestMeta {
return &activityv1.RequestMeta{
RequestId: meta.GetRequestId(),
Caller: "room-service",
GatewayNodeId: meta.GetGatewayNodeId(),
SentAtMs: time.Now().UTC().UnixMilli(),
AppCode: appcode.FromContext(ctx),
}
}
func luckyGiftDeviceID(cmd command.SendGift) string {
if value := strings.TrimSpace(cmd.SessionID); value != "" {
return value
}
return cmd.ID()
}
func luckyGiftAnchorID(roomMeta RoomMeta) string {
if roomMeta.OwnerUserID <= 0 {
return roomMeta.RoomID
}
return fmt.Sprintf("%d", roomMeta.OwnerUserID)
}
func luckyGiftResultFromProto(result *activityv1.LuckyGiftDrawResult) *roomv1.LuckyGiftDrawResult {
if result == nil {
return nil
}
return &roomv1.LuckyGiftDrawResult{
Enabled: true,
DrawId: result.GetDrawId(),
CommandId: result.GetCommandId(),
PoolId: result.GetPoolId(),
GiftId: result.GetGiftId(),
RuleVersion: result.GetRuleVersion(),
ExperiencePool: result.GetExperiencePool(),
SelectedTierId: result.GetSelectedTierId(),
MultiplierPpm: result.GetMultiplierPpm(),
BaseRewardCoins: result.GetBaseRewardCoins(),
RoomAtmosphereRewardCoins: result.GetRoomAtmosphereRewardCoins(),
ActivitySubsidyCoins: result.GetActivitySubsidyCoins(),
EffectiveRewardCoins: result.GetEffectiveRewardCoins(),
RewardStatus: result.GetRewardStatus(),
StageFeedback: result.GetStageFeedback(),
HighMultiplier: result.GetHighMultiplier(),
CreatedAtMs: result.GetCreatedAtMs(),
WalletTransactionId: result.GetWalletTransactionId(),
CoinBalanceAfter: result.GetCoinBalanceAfter(),
}
}
func normalizeGiftTargetType(raw string) string {
raw = strings.TrimSpace(raw)
switch raw {
case "", "user", "all_mic", "all_room", "couple":
return raw
default:
return raw
}
}
func normalizeGiftTargetUserIDs(targetUserID int64, targetUserIDs []int64) []int64 {
seen := make(map[int64]bool, len(targetUserIDs)+1)
ids := make([]int64, 0, len(targetUserIDs)+1)
for _, userID := range targetUserIDs {
if userID <= 0 || seen[userID] {
continue
}
seen[userID] = true
ids = append(ids, userID)
}
if len(ids) == 0 && targetUserID > 0 {
ids = append(ids, targetUserID)
}
return ids
}
func giftTargetCommandID(commandID string, targetUserID int64, targetCount int) string {
if targetCount <= 1 {
return commandID
}
// 钱包以 command_id 做幂等主键;多目标必须按接收方派生稳定子命令,才能在同一批内保存独立交易。
return fmt.Sprintf("%s:target:%d", commandID, targetUserID)
}
func giftTargetBillingsFromBatch(resp *walletv1.BatchDebitGiftResponse, targetUserIDs []int64) ([]giftTargetBilling, error) {
if resp == nil || resp.GetAggregate() == nil {
return nil, xerr.New(xerr.Unavailable, "wallet batch debit response is empty")
}
receipts := make(map[int64]*walletv1.BatchDebitGiftReceipt, len(resp.GetReceipts()))
for _, receipt := range resp.GetReceipts() {
if receipt.GetTargetUserId() <= 0 || receipt.GetBilling() == nil {
return nil, xerr.New(xerr.Unavailable, "wallet batch debit receipt is incomplete")
}
receipts[receipt.GetTargetUserId()] = receipt
}
targetBillings := make([]giftTargetBilling, 0, len(targetUserIDs))
for _, targetUserID := range targetUserIDs {
receipt := receipts[targetUserID]
if receipt == nil {
// 批量钱包响应必须覆盖每个房间目标;少一条就不能提交房间表现和 outbox。
return nil, xerr.New(xerr.Unavailable, "wallet batch debit receipt is missing")
}
targetBillings = append(targetBillings, giftTargetBilling{
TargetUserID: targetUserID,
CommandID: receipt.GetCommandId(),
Billing: receipt.GetBilling(),
})
}
return targetBillings, nil
}
func giftTargetUserIDsAttribute(targetUserIDs []int64) string {
if len(targetUserIDs) == 0 {
return ""
}
parts := make([]string, 0, len(targetUserIDs))
for _, targetUserID := range targetUserIDs {
parts = append(parts, strconv.FormatInt(targetUserID, 10))
}
return strings.Join(parts, ",")
}
func findRankItem(items []state.RankItem, userID int64) state.RankItem {
// 送礼后通常能找到发送方榜单项;找不到时返回 user_id 占位避免空事件。
for _, item := range items {
if item.UserID == userID {
return item
}
}
return state.RankItem{UserID: userID}
}