From 23376ab87fb211d17c446bb7406bea549725eeac Mon Sep 17 00:00:00 2001 From: zhx Date: Fri, 10 Jul 2026 11:25:22 +0800 Subject: [PATCH] fix: harden room gift panel aggregation --- pkg/rocketmqx/rocketmqx.go | 28 +- pkg/rocketmqx/rocketmqx_test.go | 94 ++++ .../http/roomapi/gift_panel_pipeline.go | 287 ++++++++++++ .../http/roomapi/gift_panel_pipeline_test.go | 416 ++++++++++++++++++ .../transport/http/roomapi/handler.go | 35 +- .../transport/http/roomapi/room_handler.go | 154 +++---- 6 files changed, 919 insertions(+), 95 deletions(-) create mode 100644 pkg/rocketmqx/rocketmqx_test.go create mode 100644 services/gateway-service/internal/transport/http/roomapi/gift_panel_pipeline.go create mode 100644 services/gateway-service/internal/transport/http/roomapi/gift_panel_pipeline_test.go diff --git a/pkg/rocketmqx/rocketmqx.go b/pkg/rocketmqx/rocketmqx.go index 483178f9..5de4be1f 100644 --- a/pkg/rocketmqx/rocketmqx.go +++ b/pkg/rocketmqx/rocketmqx.go @@ -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 diff --git a/pkg/rocketmqx/rocketmqx_test.go b/pkg/rocketmqx/rocketmqx_test.go new file mode 100644 index 00000000..33d789bf --- /dev/null +++ b/pkg/rocketmqx/rocketmqx_test.go @@ -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) + } + } +} diff --git a/services/gateway-service/internal/transport/http/roomapi/gift_panel_pipeline.go b/services/gateway-service/internal/transport/http/roomapi/gift_panel_pipeline.go new file mode 100644 index 00000000..f2f044b1 --- /dev/null +++ b/services/gateway-service/internal/transport/http/roomapi/gift_panel_pipeline.go @@ -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 +} diff --git a/services/gateway-service/internal/transport/http/roomapi/gift_panel_pipeline_test.go b/services/gateway-service/internal/transport/http/roomapi/gift_panel_pipeline_test.go new file mode 100644 index 00000000..a9001df8 --- /dev/null +++ b/services/gateway-service/internal/transport/http/roomapi/gift_panel_pipeline_test.go @@ -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"}, + }} +} diff --git a/services/gateway-service/internal/transport/http/roomapi/handler.go b/services/gateway-service/internal/transport/http/roomapi/handler.go index e517f26a..de4f8e3d 100644 --- a/services/gateway-service/internal/transport/http/roomapi/handler.go +++ b/services/gateway-service/internal/transport/http/roomapi/handler.go @@ -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, } } diff --git a/services/gateway-service/internal/transport/http/roomapi/room_handler.go b/services/gateway-service/internal/transport/http/roomapi/room_handler.go index 85c5409a..336d3ffa 100644 --- a/services/gateway-service/internal/transport/http/roomapi/room_handler.go +++ b/services/gateway-service/internal/transport/http/roomapi/room_handler.go @@ -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 返回房间火箭物料配置和当前进度。