fix: harden room gift panel aggregation
This commit is contained in:
zhx
parent
c09e7be659
commit
23376ab87f
@ -7,11 +7,14 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"hyapp/pkg/logx"
|
||||
|
||||
rocketmq "github.com/apache/rocketmq-client-go/v2"
|
||||
"github.com/apache/rocketmq-client-go/v2/consumer"
|
||||
"github.com/apache/rocketmq-client-go/v2/primitive"
|
||||
@ -178,6 +181,7 @@ func (p *Producer) SendSync(ctx context.Context, message Message) error {
|
||||
type Consumer struct {
|
||||
mu sync.Mutex
|
||||
client rocketmq.PushConsumer
|
||||
group string
|
||||
started bool
|
||||
}
|
||||
|
||||
@ -218,7 +222,7 @@ func NewConsumer(cfg ConsumerConfig) (*Consumer, error) {
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &Consumer{client: client}, nil
|
||||
return &Consumer{client: client, group: group}, nil
|
||||
}
|
||||
|
||||
// Subscribe binds a topic/tag selector to a handler. Empty tag subscribes all.
|
||||
@ -242,7 +246,7 @@ func (c *Consumer) Subscribe(topic string, tag string, handler Handler) error {
|
||||
if message == nil {
|
||||
continue
|
||||
}
|
||||
err := handler(ctx, ConsumedMessage{
|
||||
consumed := ConsumedMessage{
|
||||
Topic: message.Topic,
|
||||
Tag: message.GetTags(),
|
||||
Keys: message.GetKeys(),
|
||||
@ -250,9 +254,23 @@ func (c *Consumer) Subscribe(topic string, tag string, handler Handler) error {
|
||||
Body: message.Body,
|
||||
Properties: message.GetProperties(),
|
||||
ReconsumeTimes: message.ReconsumeTimes,
|
||||
})
|
||||
if err != nil {
|
||||
return consumer.ConsumeRetryLater, err
|
||||
}
|
||||
if err := handler(ctx, consumed); err != nil {
|
||||
// rocketmq-client-go v2.1.2 在 callback error 非空时会把同一 pull batch 的 MessageExt
|
||||
// 整批交给日志格式化；其 Message.String 未持有 properties 锁，而并行消费 goroutine
|
||||
// 同时写 CONSUME_START_TIME 会触发 runtime fatal，进程无法通过 recover 自救。这里仅记录
|
||||
// 已复制出的标量身份，不传 Message、Body 或 Properties，避免日志再次进入不安全 String。
|
||||
logx.Error(ctx, "rocketmq_consume_handler_failed", err,
|
||||
slog.String("consumer_group", c.group),
|
||||
slog.String("topic", consumed.Topic),
|
||||
slog.String("tag", consumed.Tag),
|
||||
slog.String("keys", consumed.Keys),
|
||||
slog.String("msg_id", consumed.MsgID),
|
||||
slog.Int("reconsume_times", int(consumed.ReconsumeTimes)),
|
||||
)
|
||||
// ConsumeRetryLater 才是 SDK 的 send-back/retry 决策信号；callback error 仅驱动其
|
||||
// 危险的整批错误日志。返回 nil error 不会确认消息，仍保留原有重试与死信策略。
|
||||
return consumer.ConsumeRetryLater, nil
|
||||
}
|
||||
}
|
||||
return consumer.ConsumeSuccess, nil
|
||||
|
||||
94
pkg/rocketmqx/rocketmqx_test.go
Normal file
94
pkg/rocketmqx/rocketmqx_test.go
Normal file
@ -0,0 +1,94 @@
|
||||
package rocketmqx
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"hyapp/pkg/logx"
|
||||
|
||||
"github.com/apache/rocketmq-client-go/v2/consumer"
|
||||
"github.com/apache/rocketmq-client-go/v2/primitive"
|
||||
)
|
||||
|
||||
type fakePushConsumer struct {
|
||||
callback func(context.Context, ...*primitive.MessageExt) (consumer.ConsumeResult, error)
|
||||
}
|
||||
|
||||
func (f *fakePushConsumer) Start() error { return nil }
|
||||
func (f *fakePushConsumer) Shutdown() error { return nil }
|
||||
func (f *fakePushConsumer) Subscribe(_ string, _ consumer.MessageSelector, callback func(context.Context, ...*primitive.MessageExt) (consumer.ConsumeResult, error)) error {
|
||||
f.callback = callback
|
||||
return nil
|
||||
}
|
||||
func (f *fakePushConsumer) Unsubscribe(string) error { return nil }
|
||||
func (f *fakePushConsumer) Suspend() {}
|
||||
func (f *fakePushConsumer) Resume() {}
|
||||
func (f *fakePushConsumer) GetOffsetDiffMap() map[string]int64 { return nil }
|
||||
|
||||
func TestSubscribeHandlerErrorReturnsRetryLaterWithoutSDKError(t *testing.T) {
|
||||
var output bytes.Buffer
|
||||
if err := logx.Init(logx.Config{Service: "rocketmqx-test", Env: "test", Format: "json", Output: &output}); err != nil {
|
||||
t.Fatalf("init test logger: %v", err)
|
||||
}
|
||||
|
||||
client := &fakePushConsumer{}
|
||||
consumerWrapper := &Consumer{client: client, group: "group-safe-log"}
|
||||
handlerErr := errors.New("projection unavailable")
|
||||
if err := consumerWrapper.Subscribe("topic-safe-log", "tag-safe-log", func(context.Context, ConsumedMessage) error {
|
||||
return handlerErr
|
||||
}); err != nil {
|
||||
t.Fatalf("subscribe: %v", err)
|
||||
}
|
||||
if client.callback == nil {
|
||||
t.Fatal("subscribe callback was not captured")
|
||||
}
|
||||
|
||||
extended := &primitive.MessageExt{MsgId: "msg-safe-log", ReconsumeTimes: 3}
|
||||
// 直接初始化嵌入的 Message，避免复制其 RWMutex；Queue 仍故意保持 nil，确保日志路径
|
||||
// 一旦把 MessageExt 交给 String 格式化就会在测试中立即暴露。
|
||||
extended.Topic = "topic-safe-log"
|
||||
extended.Body = []byte("sensitive-body-must-not-be-logged")
|
||||
extended.WithTag("tag-safe-log")
|
||||
extended.WithKeys([]string{"event-safe-log"})
|
||||
extended.WithProperty("private-property", "secret-value-must-not-be-logged")
|
||||
// Queue 故意保持 nil：MessageExt.String 会解引用 Queue 并 panic。callback 能完成并写日志，
|
||||
// 因而同时锁定包装层只记录复制后的安全标量，绝不把 MessageExt 交给格式化器。
|
||||
result, callbackErr := client.callback(context.Background(), extended)
|
||||
if result != consumer.ConsumeRetryLater {
|
||||
t.Fatalf("consume result = %v, want ConsumeRetryLater", result)
|
||||
}
|
||||
if callbackErr != nil {
|
||||
t.Fatalf("callback error = %v, want nil so SDK does not format the shared message batch", callbackErr)
|
||||
}
|
||||
|
||||
rawLog := strings.TrimSpace(output.String())
|
||||
var record map[string]any
|
||||
if err := json.Unmarshal([]byte(rawLog), &record); err != nil {
|
||||
t.Fatalf("decode safe consume log %q: %v", rawLog, err)
|
||||
}
|
||||
for key, want := range map[string]string{
|
||||
"msg": "rocketmq_consume_handler_failed",
|
||||
"consumer_group": "group-safe-log",
|
||||
"topic": "topic-safe-log",
|
||||
"tag": "tag-safe-log",
|
||||
"keys": "event-safe-log",
|
||||
"msg_id": "msg-safe-log",
|
||||
"error": handlerErr.Error(),
|
||||
} {
|
||||
if got := record[key]; got != want {
|
||||
t.Errorf("log field %s = %#v, want %q", key, got, want)
|
||||
}
|
||||
}
|
||||
if got := record["reconsume_times"]; got != float64(3) {
|
||||
t.Errorf("log field reconsume_times = %#v, want 3", got)
|
||||
}
|
||||
for _, forbidden := range []string{"sensitive-body-must-not-be-logged", "private-property", "secret-value-must-not-be-logged"} {
|
||||
if strings.Contains(rawLog, forbidden) {
|
||||
t.Errorf("safe consume log leaked forbidden message data %q: %s", forbidden, rawLog)
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,287 @@
|
||||
package roomapi
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
roomv1 "hyapp.local/api/proto/room/v1"
|
||||
userv1 "hyapp.local/api/proto/user/v1"
|
||||
walletv1 "hyapp.local/api/proto/wallet/v1"
|
||||
"hyapp/pkg/appcode"
|
||||
"hyapp/pkg/logx"
|
||||
"hyapp/pkg/roomid"
|
||||
"hyapp/services/gateway-service/internal/auth"
|
||||
"hyapp/services/gateway-service/internal/transport/http/httpkit"
|
||||
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/status"
|
||||
)
|
||||
|
||||
const (
|
||||
// Flutter 在 15 秒终止请求；gateway 留出 5 秒给 DNS/TLS/传输和客户端解码，必须先返回带 request_id 的稳定错误。
|
||||
roomGiftPanelTotalBudget = 10 * time.Second
|
||||
// 用户主资料是收礼人头像昵称的来源，单独给足正常批量查询时间，但不允许占满整个面板预算。
|
||||
roomGiftPanelRecipientCoreBudget = 1500 * time.Millisecond
|
||||
)
|
||||
|
||||
type roomGiftPanelCatalogState struct {
|
||||
version int64
|
||||
cacheable bool
|
||||
}
|
||||
|
||||
type roomGiftPanelStageResult[T any] struct {
|
||||
value T
|
||||
degraded bool
|
||||
err error
|
||||
}
|
||||
|
||||
// startRoomGiftPanelStage 只负责请求内并发和观测；业务 owner 仍由各下游服务保持。
|
||||
// 结果通道有缓冲，handler 因总预算先返回时，已响应的 gRPC goroutine 不会卡在投递结果上。
|
||||
func startRoomGiftPanelStage[T any](group *sync.WaitGroup, ctx context.Context, requestID string, roomID string, stage string, fn func() (T, bool, error)) <-chan roomGiftPanelStageResult[T] {
|
||||
resultCh := make(chan roomGiftPanelStageResult[T], 1)
|
||||
group.Add(1)
|
||||
go func() {
|
||||
defer group.Done()
|
||||
startedAt := time.Now()
|
||||
value, degraded, err := fn()
|
||||
logRoomGiftPanelStage(ctx, requestID, roomID, stage, startedAt, err, degraded)
|
||||
resultCh <- roomGiftPanelStageResult[T]{value: value, degraded: degraded, err: err}
|
||||
}()
|
||||
return resultCh
|
||||
}
|
||||
|
||||
func waitRoomGiftPanelStage[T any](ctx context.Context, resultCh <-chan roomGiftPanelStageResult[T]) (T, bool, error) {
|
||||
select {
|
||||
case result := <-resultCh:
|
||||
return result.value, result.degraded, result.err
|
||||
case <-ctx.Done():
|
||||
var zero T
|
||||
return zero, false, ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
func logRoomGiftPanelStage(ctx context.Context, requestID string, roomID string, stage string, startedAt time.Time, err error, degraded bool) {
|
||||
logCtx := logx.With(ctx,
|
||||
slog.String("request_id", requestID),
|
||||
slog.String("room_id", roomID),
|
||||
slog.String("app_code", appcode.FromContext(ctx)),
|
||||
)
|
||||
attrs := []slog.Attr{
|
||||
slog.String("stage", stage),
|
||||
slog.Int64("duration_ms", time.Since(startedAt).Milliseconds()),
|
||||
slog.Bool("success", err == nil),
|
||||
slog.Bool("degraded", degraded),
|
||||
}
|
||||
switch {
|
||||
case err != nil && degraded:
|
||||
attrs = append(attrs, slog.String("error", err.Error()))
|
||||
logx.Warn(logCtx, "gateway_room_gift_panel_stage", attrs...)
|
||||
case err != nil:
|
||||
logx.Error(logCtx, "gateway_room_gift_panel_stage", err, attrs...)
|
||||
case degraded:
|
||||
logx.Warn(logCtx, "gateway_room_gift_panel_stage", attrs...)
|
||||
default:
|
||||
logx.Info(logCtx, "gateway_room_gift_panel_stage", attrs...)
|
||||
}
|
||||
}
|
||||
|
||||
// serveRoomGiftPanel 在单个明确预算内聚合 owner service 读模型。
|
||||
// 第一阶段只并发互不依赖的 RPC；依赖 visible_region_id 的礼物配置在快照返回后启动，避免猜测区域。
|
||||
func (h *Handler) serveRoomGiftPanel(writer http.ResponseWriter, request *http.Request) {
|
||||
if h.roomQueryClient == nil || h.walletClient == nil {
|
||||
httpkit.WriteError(writer, request, http.StatusBadGateway, httpkit.CodeUpstreamError, "upstream service error")
|
||||
return
|
||||
}
|
||||
roomID := strings.TrimSpace(request.PathValue("room_id"))
|
||||
if !roomid.ValidStringID(roomID) {
|
||||
httpkit.WriteError(writer, request, http.StatusBadRequest, httpkit.CodeInvalidArgument, "invalid argument")
|
||||
return
|
||||
}
|
||||
|
||||
totalStartedAt := time.Now()
|
||||
requestID := httpkit.RequestIDFromContext(request.Context())
|
||||
totalCtx, cancel := context.WithTimeout(request.Context(), h.roomGiftPanelTotalTimeout())
|
||||
var stageGroup sync.WaitGroup
|
||||
defer func() {
|
||||
// 先取消所有尚未完成的 RPC，再等待 request-scoped worker 退出；HTTP handler 返回时不遗留后台访问。
|
||||
cancel()
|
||||
stageGroup.Wait()
|
||||
}()
|
||||
request = request.WithContext(totalCtx)
|
||||
viewerUserID := auth.UserIDFromContext(totalCtx)
|
||||
app := appcode.FromContext(totalCtx)
|
||||
|
||||
snapshotCh := startRoomGiftPanelStage(&stageGroup, totalCtx, requestID, roomID, "snapshot", func() (*roomv1.GetRoomSnapshotResponse, bool, error) {
|
||||
resp, err := h.roomQueryClient.GetRoomSnapshot(totalCtx, &roomv1.GetRoomSnapshotRequest{
|
||||
Meta: httpkit.RoomMeta(request, roomID, ""),
|
||||
RoomId: roomID,
|
||||
ViewerUserId: viewerUserID,
|
||||
})
|
||||
return resp, false, err
|
||||
})
|
||||
balanceCh := startRoomGiftPanelStage(&stageGroup, totalCtx, requestID, roomID, "balance", func() (*walletv1.GetBalancesResponse, bool, error) {
|
||||
resp, err := h.walletClient.GetBalances(totalCtx, &walletv1.GetBalancesRequest{
|
||||
RequestId: requestID,
|
||||
UserId: viewerUserID,
|
||||
AssetTypes: []string{"COIN"},
|
||||
AppCode: app,
|
||||
})
|
||||
return resp, false, err
|
||||
})
|
||||
catalogCh := startRoomGiftPanelStage(&stageGroup, totalCtx, requestID, roomID, "catalog", func() (roomGiftPanelCatalogState, bool, error) {
|
||||
version, cacheable, err := h.roomGiftCatalogVersion(request, app)
|
||||
return roomGiftPanelCatalogState{version: version, cacheable: cacheable}, false, err
|
||||
})
|
||||
bagCh := startRoomGiftPanelStage(&stageGroup, totalCtx, requestID, roomID, "bag", func() (*walletv1.ListUserResourcesResponse, bool, error) {
|
||||
resp, err := h.roomGiftBagResources(request, viewerUserID)
|
||||
return resp, false, err
|
||||
})
|
||||
|
||||
snapshotResp, _, err := waitRoomGiftPanelStage(totalCtx, snapshotCh)
|
||||
if err != nil {
|
||||
h.writeRoomGiftPanelFailure(writer, request, requestID, roomID, totalStartedAt, err)
|
||||
return
|
||||
}
|
||||
snapshot := snapshotResp.GetRoom()
|
||||
recipientUserIDs := roomGiftRecipientUserIDs(snapshot)
|
||||
profileCh := startRoomGiftPanelStage(&stageGroup, totalCtx, requestID, roomID, "recipient_profile", func() (map[int64]roomDisplayProfileData, bool, error) {
|
||||
profiles, degraded := h.roomGiftRecipientProfileMap(request, requestID, roomID, recipientUserIDs)
|
||||
return profiles, degraded, nil
|
||||
})
|
||||
|
||||
catalog, _, err := waitRoomGiftPanelStage(totalCtx, catalogCh)
|
||||
if err != nil {
|
||||
h.writeRoomGiftPanelFailure(writer, request, requestID, roomID, totalStartedAt, err)
|
||||
return
|
||||
}
|
||||
configCh := startRoomGiftPanelStage(&stageGroup, totalCtx, requestID, roomID, "config", func() (roomGiftPanelConfig, bool, error) {
|
||||
config, err := h.roomGiftPanelBaseConfigForVersion(request, app, snapshot.GetVisibleRegionId(), catalog.version, catalog.cacheable)
|
||||
return config, false, err
|
||||
})
|
||||
|
||||
balances, _, err := waitRoomGiftPanelStage(totalCtx, balanceCh)
|
||||
if err != nil {
|
||||
h.writeRoomGiftPanelFailure(writer, request, requestID, roomID, totalStartedAt, err)
|
||||
return
|
||||
}
|
||||
panelConfig, _, err := waitRoomGiftPanelStage(totalCtx, configCh)
|
||||
if err != nil {
|
||||
h.writeRoomGiftPanelFailure(writer, request, requestID, roomID, totalStartedAt, err)
|
||||
return
|
||||
}
|
||||
bagResources, _, err := waitRoomGiftPanelStage(totalCtx, bagCh)
|
||||
if err != nil {
|
||||
h.writeRoomGiftPanelFailure(writer, request, requestID, roomID, totalStartedAt, err)
|
||||
return
|
||||
}
|
||||
recipientProfiles, _, err := waitRoomGiftPanelStage(totalCtx, profileCh)
|
||||
if err != nil {
|
||||
// 核心资料的 1.5 秒子预算会在内部降级；这里出现 error 只可能是整个 gift-panel 总预算已耗尽。
|
||||
h.writeRoomGiftPanelFailure(writer, request, requestID, roomID, totalStartedAt, err)
|
||||
return
|
||||
}
|
||||
|
||||
panelGifts := append([]giftConfigData{}, panelConfig.Gifts...)
|
||||
panelGifts = append(panelGifts, roomGiftBagGiftsFromResources(bagResources, panelConfig.Gifts)...)
|
||||
data := roomGiftPanelData{
|
||||
CoinBalance: coinBalanceFromProto(balances),
|
||||
Recipients: roomGiftRecipients(snapshot, recipientProfiles),
|
||||
Tabs: roomGiftTabs(panelGifts, panelConfig.GiftTypes),
|
||||
Gifts: panelGifts,
|
||||
QuantityPresets: []int32{1, 9, 99, 999},
|
||||
}
|
||||
logRoomGiftPanelStage(totalCtx, requestID, roomID, "total", totalStartedAt, nil, false)
|
||||
httpkit.WriteOK(writer, request, data)
|
||||
}
|
||||
|
||||
func (h *Handler) writeRoomGiftPanelFailure(writer http.ResponseWriter, request *http.Request, requestID string, roomID string, startedAt time.Time, err error) {
|
||||
logRoomGiftPanelStage(request.Context(), requestID, roomID, "total", startedAt, err, false)
|
||||
if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) || status.Code(err) == codes.DeadlineExceeded || status.Code(err) == codes.Canceled || request.Context().Err() != nil {
|
||||
// 明确在 Flutter 15 秒前结束 HTTP 响应；envelope 使用原请求上下文中的 request_id，便于从客户端反查阶段日志。
|
||||
httpkit.WriteError(writer, request, http.StatusBadGateway, httpkit.CodeUpstreamError, "upstream service error")
|
||||
return
|
||||
}
|
||||
httpkit.WriteRPCError(writer, request, err)
|
||||
}
|
||||
|
||||
func (h *Handler) roomGiftPanelTotalTimeout() time.Duration {
|
||||
if h.giftPanelTotalBudget > 0 {
|
||||
return h.giftPanelTotalBudget
|
||||
}
|
||||
return roomGiftPanelTotalBudget
|
||||
}
|
||||
|
||||
func (h *Handler) roomGiftPanelRecipientCoreTimeout() time.Duration {
|
||||
if h.giftPanelRecipientCoreBudget > 0 {
|
||||
return h.giftPanelRecipientCoreBudget
|
||||
}
|
||||
return roomGiftPanelRecipientCoreBudget
|
||||
}
|
||||
|
||||
// roomGiftRecipientProfileMap 只读取 Flutter gift-panel 实际消费的头像昵称核心资料。
|
||||
// 装扮、徽章和等级字段维持空对象兼容旧 JSON 形状，但不再发起 activity/wallet enrich RPC；
|
||||
// 这些资料属于用户卡片而非送礼决策，删除整条 N+1 链路比等待后再超时降级更可靠。
|
||||
func (h *Handler) roomGiftRecipientProfileMap(request *http.Request, requestID string, roomID string, userIDs []int64) (map[int64]roomDisplayProfileData, bool) {
|
||||
userIDs = uniquePositiveUserIDs(userIDs)
|
||||
if len(userIDs) == 0 {
|
||||
return nil, false
|
||||
}
|
||||
if h.userProfileClient == nil {
|
||||
return nil, true
|
||||
}
|
||||
|
||||
coreCtx, cancelCore := context.WithTimeout(request.Context(), h.roomGiftPanelRecipientCoreTimeout())
|
||||
defer cancelCore()
|
||||
coreRequest := request.WithContext(coreCtx)
|
||||
startedAt := time.Now()
|
||||
resp, err := h.userProfileClient.BatchGetUsers(coreCtx, &userv1.BatchGetUsersRequest{
|
||||
Meta: httpkit.UserMeta(coreRequest, ""),
|
||||
UserIds: userIDs,
|
||||
})
|
||||
logRoomGiftPanelStage(coreCtx, requestID, roomID, "recipient_profile_core", startedAt, err, err != nil)
|
||||
if err != nil {
|
||||
// 房间目标 user_id 仍来自 Room Cell 快照；资料服务抖动时保留可送礼目标，只降级头像昵称。
|
||||
return nil, true
|
||||
}
|
||||
return roomGiftCoreProfilesFromUsers(userIDs, resp.GetUsers()), false
|
||||
}
|
||||
|
||||
func roomGiftCoreProfilesFromUsers(userIDs []int64, users map[int64]*userv1.User) map[int64]roomDisplayProfileData {
|
||||
profiles := make(map[int64]roomDisplayProfileData, len(users))
|
||||
for _, userID := range userIDs {
|
||||
user := users[userID]
|
||||
if user == nil {
|
||||
continue
|
||||
}
|
||||
// 核心层保留历史完整资料字段；可选资源统一初始化为空对象，超时降级不会改变 JSON 类型。
|
||||
profiles[userID] = roomDisplayProfileData{
|
||||
UserID: formatOptionalUserID(user.GetUserId()),
|
||||
Username: user.GetUsername(),
|
||||
Avatar: user.GetAvatar(),
|
||||
DisplayUserID: user.GetDisplayUserId(),
|
||||
PrettyID: user.GetPrettyId(),
|
||||
PrettyDisplayUserID: user.GetPrettyDisplayUserId(),
|
||||
Gender: strings.TrimSpace(user.GetGender()),
|
||||
Age: roomProfileAgeFromBirth(user.GetBirth(), time.Now().UTC()),
|
||||
Country: strings.ToUpper(strings.TrimSpace(user.GetCountry())),
|
||||
CountryName: strings.TrimSpace(user.GetCountryName()),
|
||||
CountryDisplayName: strings.TrimSpace(user.GetCountryDisplayName()),
|
||||
CountryFlag: roomProfileCountryFlag(user.GetCountry()),
|
||||
VIP: map[string]any{},
|
||||
Level: roomDefaultLevelDisplay(),
|
||||
LevelBadges: map[string]any{},
|
||||
Badges: []map[string]any{},
|
||||
AvatarFrame: map[string]any{},
|
||||
ProfileCard: map[string]any{},
|
||||
Vehicle: map[string]any{},
|
||||
MicSeatAnimation: map[string]any{},
|
||||
Charm: 0,
|
||||
}
|
||||
}
|
||||
return profiles
|
||||
}
|
||||
@ -0,0 +1,416 @@
|
||||
package roomapi
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
activityv1 "hyapp.local/api/proto/activity/v1"
|
||||
roomv1 "hyapp.local/api/proto/room/v1"
|
||||
userv1 "hyapp.local/api/proto/user/v1"
|
||||
walletv1 "hyapp.local/api/proto/wallet/v1"
|
||||
"hyapp/pkg/appcode"
|
||||
"hyapp/services/gateway-service/internal/auth"
|
||||
gatewayclient "hyapp/services/gateway-service/internal/client"
|
||||
"hyapp/services/gateway-service/internal/transport/http/httpkit"
|
||||
)
|
||||
|
||||
type giftPanelTestRoomQueryClient struct {
|
||||
gatewayclient.RoomQueryClient
|
||||
getSnapshot func(context.Context, *roomv1.GetRoomSnapshotRequest) (*roomv1.GetRoomSnapshotResponse, error)
|
||||
}
|
||||
|
||||
func (f *giftPanelTestRoomQueryClient) GetRoomSnapshot(ctx context.Context, req *roomv1.GetRoomSnapshotRequest) (*roomv1.GetRoomSnapshotResponse, error) {
|
||||
return f.getSnapshot(ctx, req)
|
||||
}
|
||||
|
||||
type giftPanelTestWalletClient struct {
|
||||
gatewayclient.WalletClient
|
||||
getBalances func(context.Context, *walletv1.GetBalancesRequest) (*walletv1.GetBalancesResponse, error)
|
||||
getCatalogVersion func(context.Context, *walletv1.GetGiftCatalogVersionRequest) (*walletv1.GetGiftCatalogVersionResponse, error)
|
||||
listUserResources func(context.Context, *walletv1.ListUserResourcesRequest) (*walletv1.ListUserResourcesResponse, error)
|
||||
listGiftTypes func(context.Context, *walletv1.ListGiftTypeConfigsRequest) (*walletv1.ListGiftTypeConfigsResponse, error)
|
||||
listGiftConfigs func(context.Context, *walletv1.ListGiftConfigsRequest) (*walletv1.ListGiftConfigsResponse, error)
|
||||
unusedEnrichRPCCalls atomic.Int32
|
||||
}
|
||||
|
||||
func (f *giftPanelTestWalletClient) GetBalances(ctx context.Context, req *walletv1.GetBalancesRequest) (*walletv1.GetBalancesResponse, error) {
|
||||
return f.getBalances(ctx, req)
|
||||
}
|
||||
|
||||
func (f *giftPanelTestWalletClient) GetGiftCatalogVersion(ctx context.Context, req *walletv1.GetGiftCatalogVersionRequest) (*walletv1.GetGiftCatalogVersionResponse, error) {
|
||||
return f.getCatalogVersion(ctx, req)
|
||||
}
|
||||
|
||||
func (f *giftPanelTestWalletClient) ListUserResources(ctx context.Context, req *walletv1.ListUserResourcesRequest) (*walletv1.ListUserResourcesResponse, error) {
|
||||
return f.listUserResources(ctx, req)
|
||||
}
|
||||
|
||||
func (f *giftPanelTestWalletClient) ListGiftTypeConfigs(ctx context.Context, req *walletv1.ListGiftTypeConfigsRequest) (*walletv1.ListGiftTypeConfigsResponse, error) {
|
||||
return f.listGiftTypes(ctx, req)
|
||||
}
|
||||
|
||||
func (f *giftPanelTestWalletClient) ListGiftConfigs(ctx context.Context, req *walletv1.ListGiftConfigsRequest) (*walletv1.ListGiftConfigsResponse, error) {
|
||||
return f.listGiftConfigs(ctx, req)
|
||||
}
|
||||
|
||||
func (f *giftPanelTestWalletClient) BatchGetUserEquippedResources(context.Context, *walletv1.BatchGetUserEquippedResourcesRequest) (*walletv1.BatchGetUserEquippedResourcesResponse, error) {
|
||||
f.unusedEnrichRPCCalls.Add(1)
|
||||
return &walletv1.BatchGetUserEquippedResourcesResponse{}, nil
|
||||
}
|
||||
|
||||
func (f *giftPanelTestWalletClient) GetResource(context.Context, *walletv1.GetResourceRequest) (*walletv1.GetResourceResponse, error) {
|
||||
f.unusedEnrichRPCCalls.Add(1)
|
||||
return &walletv1.GetResourceResponse{}, nil
|
||||
}
|
||||
|
||||
type giftPanelTestUserClient struct {
|
||||
gatewayclient.UserProfileClient
|
||||
batchGetUsers func(context.Context, *userv1.BatchGetUsersRequest) (*userv1.BatchGetUsersResponse, error)
|
||||
calls atomic.Int32
|
||||
}
|
||||
|
||||
func (f *giftPanelTestUserClient) BatchGetUsers(ctx context.Context, req *userv1.BatchGetUsersRequest) (*userv1.BatchGetUsersResponse, error) {
|
||||
f.calls.Add(1)
|
||||
return f.batchGetUsers(ctx, req)
|
||||
}
|
||||
|
||||
type giftPanelUnusedAchievementClient struct {
|
||||
gatewayclient.AchievementClient
|
||||
calls atomic.Int32
|
||||
}
|
||||
|
||||
func (f *giftPanelUnusedAchievementClient) ListMyBadges(context.Context, *activityv1.ListMyBadgesRequest) (*activityv1.ListMyBadgesResponse, error) {
|
||||
f.calls.Add(1)
|
||||
return &activityv1.ListMyBadgesResponse{}, nil
|
||||
}
|
||||
|
||||
type giftPanelUnusedGrowthClient struct {
|
||||
gatewayclient.GrowthLevelClient
|
||||
calls atomic.Int32
|
||||
}
|
||||
|
||||
func (f *giftPanelUnusedGrowthClient) BatchGetUserLevelDisplayProfiles(context.Context, *activityv1.BatchGetUserLevelDisplayProfilesRequest) (*activityv1.BatchGetUserLevelDisplayProfilesResponse, error) {
|
||||
f.calls.Add(1)
|
||||
return &activityv1.BatchGetUserLevelDisplayProfilesResponse{}, nil
|
||||
}
|
||||
|
||||
type giftPanelStageGate struct {
|
||||
started chan string
|
||||
release chan struct{}
|
||||
}
|
||||
|
||||
func newGiftPanelStageGate() *giftPanelStageGate {
|
||||
return &giftPanelStageGate{started: make(chan string, 16), release: make(chan struct{})}
|
||||
}
|
||||
|
||||
func (g *giftPanelStageGate) wait(ctx context.Context, stage string) error {
|
||||
select {
|
||||
case g.started <- stage:
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
select {
|
||||
case <-g.release:
|
||||
return nil
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
func awaitGiftPanelStages(t *testing.T, started <-chan string, expected ...string) {
|
||||
t.Helper()
|
||||
want := make(map[string]bool, len(expected))
|
||||
for _, stage := range expected {
|
||||
want[stage] = true
|
||||
}
|
||||
deadline := time.NewTimer(time.Second)
|
||||
defer deadline.Stop()
|
||||
for len(want) > 0 {
|
||||
select {
|
||||
case stage := <-started:
|
||||
delete(want, stage)
|
||||
case <-deadline.C:
|
||||
t.Fatalf("gift-panel stages did not start concurrently; missing=%v", want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
type giftPanelBlockingTracker struct {
|
||||
active atomic.Int32
|
||||
canceled atomic.Int32
|
||||
}
|
||||
|
||||
func (t *giftPanelBlockingTracker) wait(ctx context.Context) error {
|
||||
t.active.Add(1)
|
||||
defer t.active.Add(-1)
|
||||
<-ctx.Done()
|
||||
t.canceled.Add(1)
|
||||
return ctx.Err()
|
||||
}
|
||||
|
||||
func TestRoomGiftPanelProductionBudgetsStayBelowFlutterTimeout(t *testing.T) {
|
||||
handler := New(Config{})
|
||||
if got := handler.roomGiftPanelTotalTimeout(); got != 10*time.Second || got >= 15*time.Second {
|
||||
t.Fatalf("gift-panel total budget must leave client transport headroom: %s", got)
|
||||
}
|
||||
if got := handler.roomGiftPanelRecipientCoreTimeout(); got != 1500*time.Millisecond || got >= handler.roomGiftPanelTotalTimeout() {
|
||||
t.Fatalf("recipient core budget must remain a short degradable stage: %s", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRoomGiftPanelRunsIndependentStagesConcurrentlyAndSkipsUnusedEnrichment(t *testing.T) {
|
||||
phaseOne := newGiftPanelStageGate()
|
||||
phaseTwo := newGiftPanelStageGate()
|
||||
roomID := "lalu_room_gift_parallel"
|
||||
queryClient := &giftPanelTestRoomQueryClient{getSnapshot: func(ctx context.Context, _ *roomv1.GetRoomSnapshotRequest) (*roomv1.GetRoomSnapshotResponse, error) {
|
||||
if err := phaseOne.wait(ctx, "snapshot"); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return giftPanelTestSnapshot(roomID), nil
|
||||
}}
|
||||
walletClient := giftPanelWorkingWalletClient()
|
||||
walletClient.getBalances = func(ctx context.Context, _ *walletv1.GetBalancesRequest) (*walletv1.GetBalancesResponse, error) {
|
||||
if err := phaseOne.wait(ctx, "balance"); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &walletv1.GetBalancesResponse{Balances: []*walletv1.AssetBalance{{AssetType: "COIN", AvailableAmount: 1000}}}, nil
|
||||
}
|
||||
walletClient.getCatalogVersion = func(ctx context.Context, _ *walletv1.GetGiftCatalogVersionRequest) (*walletv1.GetGiftCatalogVersionResponse, error) {
|
||||
if err := phaseOne.wait(ctx, "catalog"); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &walletv1.GetGiftCatalogVersionResponse{Version: 1}, nil
|
||||
}
|
||||
walletClient.listUserResources = func(ctx context.Context, _ *walletv1.ListUserResourcesRequest) (*walletv1.ListUserResourcesResponse, error) {
|
||||
if err := phaseOne.wait(ctx, "bag"); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &walletv1.ListUserResourcesResponse{}, nil
|
||||
}
|
||||
walletClient.listGiftTypes = func(ctx context.Context, _ *walletv1.ListGiftTypeConfigsRequest) (*walletv1.ListGiftTypeConfigsResponse, error) {
|
||||
if err := phaseTwo.wait(ctx, "gift_types"); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &walletv1.ListGiftTypeConfigsResponse{GiftTypes: []*walletv1.GiftTypeConfig{{TypeCode: "normal", TabKey: "Gift", Status: "active"}}}, nil
|
||||
}
|
||||
walletClient.listGiftConfigs = func(ctx context.Context, _ *walletv1.ListGiftConfigsRequest) (*walletv1.ListGiftConfigsResponse, error) {
|
||||
if err := phaseTwo.wait(ctx, "gifts"); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return giftPanelTestGiftConfigs(), nil
|
||||
}
|
||||
userClient := &giftPanelTestUserClient{batchGetUsers: func(ctx context.Context, _ *userv1.BatchGetUsersRequest) (*userv1.BatchGetUsersResponse, error) {
|
||||
if err := phaseTwo.wait(ctx, "recipient_core"); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return giftPanelTestUsers(), nil
|
||||
}}
|
||||
achievementClient := &giftPanelUnusedAchievementClient{}
|
||||
growthClient := &giftPanelUnusedGrowthClient{}
|
||||
handler := New(Config{
|
||||
RoomQueryClient: queryClient,
|
||||
WalletClient: walletClient,
|
||||
UserProfileClient: userClient,
|
||||
AchievementClient: achievementClient,
|
||||
GrowthLevelClient: growthClient,
|
||||
})
|
||||
handler.giftPanelRecipientCoreBudget = 3 * time.Second
|
||||
|
||||
responseCh := make(chan *httptest.ResponseRecorder, 1)
|
||||
go func() { responseCh <- serveGiftPanelForTest(handler, roomID) }()
|
||||
awaitGiftPanelStages(t, phaseOne.started, "snapshot", "balance", "catalog", "bag")
|
||||
close(phaseOne.release)
|
||||
awaitGiftPanelStages(t, phaseTwo.started, "gift_types", "gifts", "recipient_core")
|
||||
close(phaseTwo.release)
|
||||
|
||||
recorder := <-responseCh
|
||||
if recorder.Code != http.StatusOK {
|
||||
t.Fatalf("gift-panel parallel response mismatch: status=%d body=%s", recorder.Code, recorder.Body.String())
|
||||
}
|
||||
if userClient.calls.Load() != 1 {
|
||||
t.Fatalf("recipient core must use one BatchGetUsers call, got=%d", userClient.calls.Load())
|
||||
}
|
||||
if achievementClient.calls.Load() != 0 || growthClient.calls.Load() != 0 || walletClient.unusedEnrichRPCCalls.Load() != 0 {
|
||||
t.Fatalf("gift-panel must not call unused enrich RPCs: badge=%d level=%d wallet=%d", achievementClient.calls.Load(), growthClient.calls.Load(), walletClient.unusedEnrichRPCCalls.Load())
|
||||
}
|
||||
var envelope struct {
|
||||
Code string `json:"code"`
|
||||
Data struct {
|
||||
Recipients []roomGiftRecipientData `json:"recipients"`
|
||||
} `json:"data"`
|
||||
}
|
||||
if err := json.NewDecoder(recorder.Body).Decode(&envelope); err != nil {
|
||||
t.Fatalf("decode gift-panel response: %v", err)
|
||||
}
|
||||
if envelope.Code != httpkit.CodeOK || len(envelope.Data.Recipients) != 3 {
|
||||
t.Fatalf("gift-panel recipients mismatch: %+v", envelope)
|
||||
}
|
||||
profile := envelope.Data.Recipients[1].Profile
|
||||
if profile == nil || profile.Username != "Host 101" || profile.Avatar != "https://cdn.example/101.png" {
|
||||
t.Fatalf("gift-panel core profile must survive without enrich: %+v", profile)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRoomGiftPanelTotalBudgetReturnsTraceableUpstreamErrorAndStopsWorkers(t *testing.T) {
|
||||
tracker := &giftPanelBlockingTracker{}
|
||||
queryClient := &giftPanelTestRoomQueryClient{getSnapshot: func(ctx context.Context, _ *roomv1.GetRoomSnapshotRequest) (*roomv1.GetRoomSnapshotResponse, error) {
|
||||
return nil, tracker.wait(ctx)
|
||||
}}
|
||||
walletClient := giftPanelWorkingWalletClient()
|
||||
walletClient.getBalances = func(ctx context.Context, _ *walletv1.GetBalancesRequest) (*walletv1.GetBalancesResponse, error) {
|
||||
return nil, tracker.wait(ctx)
|
||||
}
|
||||
walletClient.getCatalogVersion = func(ctx context.Context, _ *walletv1.GetGiftCatalogVersionRequest) (*walletv1.GetGiftCatalogVersionResponse, error) {
|
||||
return nil, tracker.wait(ctx)
|
||||
}
|
||||
walletClient.listUserResources = func(ctx context.Context, _ *walletv1.ListUserResourcesRequest) (*walletv1.ListUserResourcesResponse, error) {
|
||||
return nil, tracker.wait(ctx)
|
||||
}
|
||||
handler := New(Config{RoomQueryClient: queryClient, WalletClient: walletClient})
|
||||
handler.giftPanelTotalBudget = 40 * time.Millisecond
|
||||
|
||||
startedAt := time.Now()
|
||||
recorder := serveGiftPanelForTest(handler, "lalu_room_gift_timeout")
|
||||
if elapsed := time.Since(startedAt); elapsed > 500*time.Millisecond {
|
||||
t.Fatalf("gift-panel budget did not stop response promptly: elapsed=%s", elapsed)
|
||||
}
|
||||
if recorder.Code != http.StatusBadGateway {
|
||||
t.Fatalf("gift-panel timeout status mismatch: got=%d body=%s", recorder.Code, recorder.Body.String())
|
||||
}
|
||||
var envelope httpkit.ResponseEnvelope
|
||||
if err := json.NewDecoder(recorder.Body).Decode(&envelope); err != nil {
|
||||
t.Fatalf("decode timeout envelope: %v", err)
|
||||
}
|
||||
if envelope.Code != httpkit.CodeUpstreamError || envelope.RequestID == "" || envelope.RequestID != recorder.Header().Get("X-Request-ID") {
|
||||
t.Fatalf("timeout envelope must expose server request_id: envelope=%+v header=%q", envelope, recorder.Header().Get("X-Request-ID"))
|
||||
}
|
||||
if tracker.canceled.Load() != 4 || tracker.active.Load() != 0 {
|
||||
t.Fatalf("all request workers must observe cancellation and exit: canceled=%d active=%d", tracker.canceled.Load(), tracker.active.Load())
|
||||
}
|
||||
}
|
||||
|
||||
func TestRoomGiftPanelRecipientCoreTimeoutDegradesWithoutFailingPanel(t *testing.T) {
|
||||
tracker := &giftPanelBlockingTracker{}
|
||||
handler := New(Config{
|
||||
RoomQueryClient: &giftPanelTestRoomQueryClient{getSnapshot: func(context.Context, *roomv1.GetRoomSnapshotRequest) (*roomv1.GetRoomSnapshotResponse, error) {
|
||||
return giftPanelTestSnapshot("lalu_room_gift_profile_timeout"), nil
|
||||
}},
|
||||
WalletClient: giftPanelWorkingWalletClient(),
|
||||
UserProfileClient: &giftPanelTestUserClient{batchGetUsers: func(ctx context.Context, _ *userv1.BatchGetUsersRequest) (*userv1.BatchGetUsersResponse, error) {
|
||||
return nil, tracker.wait(ctx)
|
||||
}},
|
||||
})
|
||||
handler.giftPanelRecipientCoreBudget = 35 * time.Millisecond
|
||||
|
||||
recorder := serveGiftPanelForTest(handler, "lalu_room_gift_profile_timeout")
|
||||
if recorder.Code != http.StatusOK {
|
||||
t.Fatalf("recipient core timeout must degrade, not fail panel: status=%d body=%s", recorder.Code, recorder.Body.String())
|
||||
}
|
||||
if tracker.canceled.Load() != 1 || tracker.active.Load() != 0 {
|
||||
t.Fatalf("recipient core worker must stop at its child deadline: canceled=%d active=%d", tracker.canceled.Load(), tracker.active.Load())
|
||||
}
|
||||
var envelope struct {
|
||||
Code string `json:"code"`
|
||||
Data struct {
|
||||
Recipients []roomGiftRecipientData `json:"recipients"`
|
||||
} `json:"data"`
|
||||
}
|
||||
if err := json.NewDecoder(recorder.Body).Decode(&envelope); err != nil {
|
||||
t.Fatalf("decode degraded panel: %v", err)
|
||||
}
|
||||
if envelope.Code != httpkit.CodeOK || len(envelope.Data.Recipients) != 3 || envelope.Data.Recipients[1].UserID == "" || envelope.Data.Recipients[1].Profile != nil {
|
||||
t.Fatalf("degraded panel must preserve Room Cell recipient IDs only: %+v", envelope)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRoomGiftPanelConfigFailureCancelsSiblingRPCBeforeReturning(t *testing.T) {
|
||||
tracker := &giftPanelBlockingTracker{}
|
||||
giftStarted := make(chan struct{})
|
||||
walletClient := giftPanelWorkingWalletClient()
|
||||
walletClient.listGiftConfigs = func(ctx context.Context, _ *walletv1.ListGiftConfigsRequest) (*walletv1.ListGiftConfigsResponse, error) {
|
||||
close(giftStarted)
|
||||
return nil, tracker.wait(ctx)
|
||||
}
|
||||
walletClient.listGiftTypes = func(context.Context, *walletv1.ListGiftTypeConfigsRequest) (*walletv1.ListGiftTypeConfigsResponse, error) {
|
||||
<-giftStarted
|
||||
return nil, errors.New("gift type test failure")
|
||||
}
|
||||
handler := New(Config{WalletClient: walletClient})
|
||||
request := httptest.NewRequest(http.MethodGet, "/api/v1/rooms/lalu_room_gift_config_failure/gift-panel", nil)
|
||||
request.SetPathValue("room_id", "lalu_room_gift_config_failure")
|
||||
request = request.WithContext(appcode.WithContext(request.Context(), "lalu"))
|
||||
|
||||
_, err := handler.roomGiftPanelBaseConfigForVersion(request, "lalu", 1001, 1, false)
|
||||
if err == nil || err.Error() != "gift type test failure" {
|
||||
t.Fatalf("config error mismatch: %v", err)
|
||||
}
|
||||
if tracker.canceled.Load() != 1 || tracker.active.Load() != 0 {
|
||||
t.Fatalf("failed config sibling must observe cancellation before return: canceled=%d active=%d", tracker.canceled.Load(), tracker.active.Load())
|
||||
}
|
||||
}
|
||||
|
||||
func serveGiftPanelForTest(handler *Handler, roomID string) *httptest.ResponseRecorder {
|
||||
request := httptest.NewRequest(http.MethodGet, "/api/v1/rooms/"+roomID+"/gift-panel", nil)
|
||||
request.SetPathValue("room_id", roomID)
|
||||
request = request.WithContext(auth.WithUserID(appcode.WithContext(request.Context(), "lalu"), 42))
|
||||
recorder := httptest.NewRecorder()
|
||||
httpkit.WithRequestID(http.HandlerFunc(handler.getRoomGiftPanel)).ServeHTTP(recorder, request)
|
||||
return recorder
|
||||
}
|
||||
|
||||
func giftPanelTestSnapshot(roomID string) *roomv1.GetRoomSnapshotResponse {
|
||||
return &roomv1.GetRoomSnapshotResponse{Room: &roomv1.RoomSnapshot{
|
||||
RoomId: roomID,
|
||||
OwnerUserId: 101,
|
||||
VisibleRegionId: 1001,
|
||||
MicSeats: []*roomv1.SeatState{
|
||||
{SeatNo: 1, UserId: 101},
|
||||
{SeatNo: 2, UserId: 102},
|
||||
},
|
||||
}}
|
||||
}
|
||||
|
||||
func giftPanelWorkingWalletClient() *giftPanelTestWalletClient {
|
||||
return &giftPanelTestWalletClient{
|
||||
getBalances: func(context.Context, *walletv1.GetBalancesRequest) (*walletv1.GetBalancesResponse, error) {
|
||||
return &walletv1.GetBalancesResponse{Balances: []*walletv1.AssetBalance{{AssetType: "COIN", AvailableAmount: 1000}}}, nil
|
||||
},
|
||||
getCatalogVersion: func(context.Context, *walletv1.GetGiftCatalogVersionRequest) (*walletv1.GetGiftCatalogVersionResponse, error) {
|
||||
return &walletv1.GetGiftCatalogVersionResponse{Version: 1}, nil
|
||||
},
|
||||
listUserResources: func(context.Context, *walletv1.ListUserResourcesRequest) (*walletv1.ListUserResourcesResponse, error) {
|
||||
return &walletv1.ListUserResourcesResponse{}, nil
|
||||
},
|
||||
listGiftTypes: func(context.Context, *walletv1.ListGiftTypeConfigsRequest) (*walletv1.ListGiftTypeConfigsResponse, error) {
|
||||
return &walletv1.ListGiftTypeConfigsResponse{GiftTypes: []*walletv1.GiftTypeConfig{{TypeCode: "normal", TabKey: "Gift", Status: "active"}}}, nil
|
||||
},
|
||||
listGiftConfigs: func(context.Context, *walletv1.ListGiftConfigsRequest) (*walletv1.ListGiftConfigsResponse, error) {
|
||||
return giftPanelTestGiftConfigs(), nil
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func giftPanelTestGiftConfigs() *walletv1.ListGiftConfigsResponse {
|
||||
return &walletv1.ListGiftConfigsResponse{Gifts: []*walletv1.GiftConfig{{
|
||||
GiftId: "rose",
|
||||
ResourceId: 11,
|
||||
Resource: &walletv1.Resource{ResourceId: 11, ResourceType: "gift", Status: "active", Name: "Rose"},
|
||||
Status: "active",
|
||||
Name: "Rose",
|
||||
GiftTypeCode: "normal",
|
||||
CoinPrice: 100,
|
||||
}}}
|
||||
}
|
||||
|
||||
func giftPanelTestUsers() *userv1.BatchGetUsersResponse {
|
||||
return &userv1.BatchGetUsersResponse{Users: map[int64]*userv1.User{
|
||||
101: {UserId: 101, Username: "Host 101", Avatar: "https://cdn.example/101.png", DisplayUserId: "100101"},
|
||||
102: {UserId: 102, Username: "Host 102", Avatar: "https://cdn.example/102.png", DisplayUserId: "100102"},
|
||||
}}
|
||||
}
|
||||
@ -27,6 +27,11 @@ type Handler struct {
|
||||
growthLevelClient client.GrowthLevelClient
|
||||
achievementClient client.AchievementClient
|
||||
rtcTokenConfig tencentrtc.TokenConfig
|
||||
|
||||
// gift-panel 的预算保留在 handler 内，生产使用固定安全值；测试可缩短预算验证真实超时分支，
|
||||
// 避免把 Flutter 的 15 秒客户端超时当成后端取消机制。
|
||||
giftPanelTotalBudget time.Duration
|
||||
giftPanelRecipientCoreBudget time.Duration
|
||||
}
|
||||
|
||||
type Config struct {
|
||||
@ -47,20 +52,22 @@ type Config struct {
|
||||
|
||||
func New(config Config) *Handler {
|
||||
return &Handler{
|
||||
roomClient: config.RoomClient,
|
||||
roomGuardClient: config.RoomGuardClient,
|
||||
roomQueryClient: config.RoomQueryClient,
|
||||
userProfileClient: config.UserProfileClient,
|
||||
userCountryClient: config.UserCountryClient,
|
||||
userRegionClient: config.UserRegionClient,
|
||||
userSocialClient: config.UserSocialClient,
|
||||
userHostClient: config.UserHostClient,
|
||||
appConfigReader: config.AppConfigReader,
|
||||
walletClient: config.WalletClient,
|
||||
giftPanelCache: newRoomGiftPanelConfigCache(45 * time.Second),
|
||||
growthLevelClient: config.GrowthLevelClient,
|
||||
achievementClient: config.AchievementClient,
|
||||
rtcTokenConfig: config.RTCTokenConfig,
|
||||
roomClient: config.RoomClient,
|
||||
roomGuardClient: config.RoomGuardClient,
|
||||
roomQueryClient: config.RoomQueryClient,
|
||||
userProfileClient: config.UserProfileClient,
|
||||
userCountryClient: config.UserCountryClient,
|
||||
userRegionClient: config.UserRegionClient,
|
||||
userSocialClient: config.UserSocialClient,
|
||||
userHostClient: config.UserHostClient,
|
||||
appConfigReader: config.AppConfigReader,
|
||||
walletClient: config.WalletClient,
|
||||
giftPanelCache: newRoomGiftPanelConfigCache(45 * time.Second),
|
||||
growthLevelClient: config.GrowthLevelClient,
|
||||
achievementClient: config.AchievementClient,
|
||||
rtcTokenConfig: config.RTCTokenConfig,
|
||||
giftPanelTotalBudget: roomGiftPanelTotalBudget,
|
||||
giftPanelRecipientCoreBudget: roomGiftPanelRecipientCoreBudget,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -1,6 +1,7 @@
|
||||
package roomapi
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
@ -8,6 +9,7 @@ import (
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"hyapp/pkg/appcode"
|
||||
@ -723,59 +725,7 @@ func (h *Handler) getRoomContributionRank(writer http.ResponseWriter, request *h
|
||||
|
||||
// getRoomGiftPanel 返回房间送礼面板初始化数据。
|
||||
func (h *Handler) getRoomGiftPanel(writer http.ResponseWriter, request *http.Request) {
|
||||
if h.roomQueryClient == nil || h.walletClient == nil {
|
||||
httpkit.WriteError(writer, request, http.StatusBadGateway, httpkit.CodeUpstreamError, "upstream service error")
|
||||
return
|
||||
}
|
||||
roomID := strings.TrimSpace(request.PathValue("room_id"))
|
||||
if !roomid.ValidStringID(roomID) {
|
||||
httpkit.WriteError(writer, request, http.StatusBadRequest, httpkit.CodeInvalidArgument, "invalid argument")
|
||||
return
|
||||
}
|
||||
|
||||
viewerUserID := auth.UserIDFromContext(request.Context())
|
||||
snapshotResp, err := h.roomQueryClient.GetRoomSnapshot(request.Context(), &roomv1.GetRoomSnapshotRequest{
|
||||
Meta: httpkit.RoomMeta(request, roomID, ""),
|
||||
RoomId: roomID,
|
||||
ViewerUserId: viewerUserID,
|
||||
})
|
||||
if err != nil {
|
||||
httpkit.WriteRPCError(writer, request, err)
|
||||
return
|
||||
}
|
||||
snapshot := snapshotResp.GetRoom()
|
||||
balances, err := h.walletClient.GetBalances(request.Context(), &walletv1.GetBalancesRequest{
|
||||
RequestId: httpkit.RequestIDFromContext(request.Context()),
|
||||
UserId: viewerUserID,
|
||||
AssetTypes: []string{"COIN"},
|
||||
AppCode: appcode.FromContext(request.Context()),
|
||||
})
|
||||
if err != nil {
|
||||
httpkit.WriteRPCError(writer, request, err)
|
||||
return
|
||||
}
|
||||
panelConfig, err := h.roomGiftPanelBaseConfig(request, snapshot.GetVisibleRegionId())
|
||||
if err != nil {
|
||||
httpkit.WriteRPCError(writer, request, err)
|
||||
return
|
||||
}
|
||||
giftTypes := panelConfig.GiftTypes
|
||||
gifts := panelConfig.Gifts
|
||||
bagGifts, err := h.roomGiftBagGifts(request, viewerUserID, gifts)
|
||||
if err != nil {
|
||||
httpkit.WriteRPCError(writer, request, err)
|
||||
return
|
||||
}
|
||||
panelGifts := append([]giftConfigData{}, gifts...)
|
||||
panelGifts = append(panelGifts, bagGifts...)
|
||||
recipientProfiles := h.roomDisplayProfileMap(request, roomGiftRecipientUserIDs(snapshot))
|
||||
httpkit.WriteOK(writer, request, roomGiftPanelData{
|
||||
CoinBalance: coinBalanceFromProto(balances),
|
||||
Recipients: roomGiftRecipients(snapshot, recipientProfiles),
|
||||
Tabs: roomGiftTabs(panelGifts, giftTypes),
|
||||
Gifts: panelGifts,
|
||||
QuantityPresets: []int32{1, 9, 99, 999},
|
||||
})
|
||||
h.serveRoomGiftPanel(writer, request)
|
||||
}
|
||||
|
||||
func (h *Handler) roomGiftPanelBaseConfig(request *http.Request, regionID int64) (roomGiftPanelConfig, error) {
|
||||
@ -784,35 +734,73 @@ func (h *Handler) roomGiftPanelBaseConfig(request *http.Request, regionID int64)
|
||||
if err != nil {
|
||||
return roomGiftPanelConfig{}, err
|
||||
}
|
||||
return h.roomGiftPanelBaseConfigForVersion(request, app, regionID, catalogVersion, cacheable)
|
||||
}
|
||||
|
||||
// roomGiftPanelBaseConfigForVersion 使用已并行取得的 owner 版本加载稳定配置。
|
||||
// 类型与礼物列表互不依赖，因此 cache miss 时同时查询；区域过滤仍只由 wallet-service 解释。
|
||||
func (h *Handler) roomGiftPanelBaseConfigForVersion(request *http.Request, app string, regionID int64, catalogVersion int64, cacheable bool) (roomGiftPanelConfig, error) {
|
||||
requestID := httpkit.RequestIDFromContext(request.Context())
|
||||
roomID := strings.TrimSpace(request.PathValue("room_id"))
|
||||
if cacheable && h.giftPanelCache != nil {
|
||||
cacheStartedAt := time.Now()
|
||||
if cached, ok := h.giftPanelCache.Get(app, regionID, catalogVersion, time.Now()); ok {
|
||||
logRoomGiftPanelStage(request.Context(), requestID, roomID, "config_cache_hit", cacheStartedAt, nil, false)
|
||||
return cached, nil
|
||||
}
|
||||
}
|
||||
giftTypeResp, err := h.walletClient.ListGiftTypeConfigs(request.Context(), &walletv1.ListGiftTypeConfigsRequest{
|
||||
RequestId: httpkit.RequestIDFromContext(request.Context()),
|
||||
AppCode: app,
|
||||
ActiveOnly: true,
|
||||
})
|
||||
if err != nil {
|
||||
return roomGiftPanelConfig{}, err
|
||||
|
||||
// 任一配置 RPC 失败就取消同批 sibling，并在 Wait 后返回；这样总预算取消时不会遗留继续占连接的查询 goroutine。
|
||||
configCtx, cancelConfig := context.WithCancel(request.Context())
|
||||
configRequest := request.WithContext(configCtx)
|
||||
var waitGroup sync.WaitGroup
|
||||
var giftTypeResp *walletv1.ListGiftTypeConfigsResponse
|
||||
var giftTypeErr error
|
||||
var giftResp *walletv1.ListGiftConfigsResponse
|
||||
var giftErr error
|
||||
waitGroup.Add(2)
|
||||
go func() {
|
||||
defer waitGroup.Done()
|
||||
startedAt := time.Now()
|
||||
giftTypeResp, giftTypeErr = h.walletClient.ListGiftTypeConfigs(configCtx, &walletv1.ListGiftTypeConfigsRequest{
|
||||
RequestId: requestID,
|
||||
AppCode: app,
|
||||
ActiveOnly: true,
|
||||
})
|
||||
logRoomGiftPanelStage(configRequest.Context(), requestID, roomID, "gift_types", startedAt, giftTypeErr, false)
|
||||
if giftTypeErr != nil {
|
||||
cancelConfig()
|
||||
}
|
||||
}()
|
||||
go func() {
|
||||
defer waitGroup.Done()
|
||||
startedAt := time.Now()
|
||||
giftResp, giftErr = h.walletClient.ListGiftConfigs(configCtx, &walletv1.ListGiftConfigsRequest{
|
||||
RequestId: requestID,
|
||||
AppCode: app,
|
||||
Page: 1,
|
||||
PageSize: 500,
|
||||
ActiveOnly: true,
|
||||
RegionId: regionID,
|
||||
FilterRegion: true,
|
||||
})
|
||||
logRoomGiftPanelStage(configRequest.Context(), requestID, roomID, "gifts", startedAt, giftErr, false)
|
||||
if giftErr != nil {
|
||||
cancelConfig()
|
||||
}
|
||||
}()
|
||||
waitGroup.Wait()
|
||||
cancelConfig()
|
||||
if giftTypeErr != nil {
|
||||
return roomGiftPanelConfig{}, giftTypeErr
|
||||
}
|
||||
if giftErr != nil {
|
||||
return roomGiftPanelConfig{}, giftErr
|
||||
}
|
||||
giftTypes := make([]giftTypeConfigData, 0, len(giftTypeResp.GetGiftTypes()))
|
||||
for _, item := range giftTypeResp.GetGiftTypes() {
|
||||
giftTypes = append(giftTypes, giftTypeFromProto(item))
|
||||
}
|
||||
giftResp, err := h.walletClient.ListGiftConfigs(request.Context(), &walletv1.ListGiftConfigsRequest{
|
||||
RequestId: httpkit.RequestIDFromContext(request.Context()),
|
||||
AppCode: app,
|
||||
Page: 1,
|
||||
PageSize: 500,
|
||||
ActiveOnly: true,
|
||||
RegionId: regionID,
|
||||
FilterRegion: true,
|
||||
})
|
||||
if err != nil {
|
||||
return roomGiftPanelConfig{}, err
|
||||
}
|
||||
gifts := make([]giftConfigData, 0, len(giftResp.GetGifts()))
|
||||
for _, gift := range giftResp.GetGifts() {
|
||||
gifts = append(gifts, giftFromProto(gift))
|
||||
@ -848,15 +836,29 @@ func (h *Handler) roomGiftBagGifts(request *http.Request, viewerUserID int64, gi
|
||||
if viewerUserID <= 0 || h.walletClient == nil || len(gifts) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
resp, err := h.walletClient.ListUserResources(request.Context(), &walletv1.ListUserResourcesRequest{
|
||||
resp, err := h.roomGiftBagResources(request, viewerUserID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return roomGiftBagGiftsFromResources(resp, gifts), nil
|
||||
}
|
||||
|
||||
func (h *Handler) roomGiftBagResources(request *http.Request, viewerUserID int64) (*walletv1.ListUserResourcesResponse, error) {
|
||||
if viewerUserID <= 0 || h.walletClient == nil {
|
||||
return &walletv1.ListUserResourcesResponse{}, nil
|
||||
}
|
||||
return h.walletClient.ListUserResources(request.Context(), &walletv1.ListUserResourcesRequest{
|
||||
RequestId: httpkit.RequestIDFromContext(request.Context()),
|
||||
AppCode: appcode.FromContext(request.Context()),
|
||||
UserId: viewerUserID,
|
||||
ResourceType: "gift",
|
||||
ActiveOnly: true,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
func roomGiftBagGiftsFromResources(resp *walletv1.ListUserResourcesResponse, gifts []giftConfigData) []giftConfigData {
|
||||
if resp == nil || len(gifts) == 0 {
|
||||
return nil
|
||||
}
|
||||
giftByResourceID := make(map[int64]giftConfigData, len(gifts))
|
||||
for _, gift := range gifts {
|
||||
@ -885,7 +887,7 @@ func (h *Handler) roomGiftBagGifts(request *http.Request, viewerUserID int64, gi
|
||||
gift.RemainingQuantity = remaining
|
||||
bagGifts = append(bagGifts, gift)
|
||||
}
|
||||
return bagGifts, nil
|
||||
return bagGifts
|
||||
}
|
||||
|
||||
// getRoomRocket 返回房间火箭物料配置和当前进度。
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user