字段补充

2026-06-01 19:42:53 +08:00 · 2026-06-01 19:42:53 +08:00 · 6d35f421c0
commit 6d35f421c0
parent 05881934ad
4 changed files with 141 additions and 9 deletions
--- a/docs/flutter对接/语音房红包Flutter对接.md
+++ b/docs/flutter对接/语音房红包Flutter对接.md
@ -205,6 +205,9 @@ GET /api/v1/app/voice-room/red-packet/active?roomId=9001
        "roomId": "9001",
        "regionId": 310,
        "senderUserId": 10001,
+        "senderAccount": "10001",
+        "senderNickname": "Tom",
+        "senderAvatar": "https://example.com/avatar1.png",
        "packetMode": "DELAYED",
        "totalAmount": 15000,
        "totalCount": 5,
@ -585,7 +588,7 @@ class VoiceRoomRedPacket {
 解析建议：

 - 对 `senderUserId/userId` 这类 ID，兼容 number 和 string，统一转 string。
- HTTP 发送成功、详情、领取成功里的 `packet` 使用 `senderAccount/senderNickname/senderAvatar`；IM `CREATED/OPEN_DUE` 使用 `actualAccount/userNickname/userAvatar`，落本地模型时归一到发送者字段。
+- HTTP 发送成功、active、详情、领取成功里的 `packet` 使用 `senderAccount/senderNickname/senderAvatar`；IM `CREATED/OPEN_DUE` 使用 `actualAccount/userNickname/userAvatar`，落本地模型时归一到发送者字段。
 - 对金额和时间，兼容 number 和 string，统一转 int。
 - `CREATED` IM 里的 `remainAmount/remainCount` 对应 HTTP 里的 `remainingAmount/remainingCount`。
 - 如果 IM 和 HTTP 字段冲突，以 HTTP 最新查询结果为准。
--- a/services/activity-service/internal/service/broadcast/service_test.go
+++ b/services/activity-service/internal/service/broadcast/service_test.go
@ -154,6 +154,82 @@ func TestConsumeRedPacketCreatedBuildsFullIMPayload(t *testing.T) {
 	}
 }

+func TestDelayedRedPacketCreatedSendsRegionAndSchedulesOpenDueIM(t *testing.T) {
+	nowMS := int64(1_779_999_700_000)
+	openAtMS := int64(1_780_000_000_000)
+	repository := newFakeRepository()
+	publisher := &fakePublisher{}
+	service := New(Config{NodeID: "node-a"}, repository, publisher, nil)
+	service.SetClock(func() time.Time { return time.UnixMilli(nowMS) })
+	service.SetSenderProfileSource(fakeSenderProfileSource{profile: SenderProfile{
+		UserID:     10001,
+		Account:    "10001",
+		Nickname:   "Tom",
+		Avatar:     "https://example.com/avatar.png",
+		RegionCode: "AR",
+	}})
+	event := broadcastdomain.RedPacketWalletEvent{
+		AppCode:         "lalu",
+		EventID:         "wallet-red-packet-created-1",
+		EventType:       "WalletRedPacketCreated",
+		PacketID:        "VRRP2026052712340001",
+		PacketType:      "delayed",
+		RoomID:          "9001",
+		RegionID:        310,
+		SenderUserID:    10001,
+		TotalAmount:     15000,
+		TotalCount:      5,
+		RemainingAmount: 15000,
+		RemainingCount:  5,
+		Status:          "waiting_open",
+		OpenAtMS:        openAtMS,
+		ExpiresAtMS:     1_780_086_400_000,
+		CreatedAtMS:     nowMS,
+	}
+
+	if err := service.ConsumeRedPacketWalletEvent(appcode.WithContext(context.Background(), "lalu"), event); err != nil {
+		t.Fatalf("ConsumeRedPacketWalletEvent failed: %v", err)
+	}
+
+	roomCreated := repository.records[event.EventID]
+	regionCreated := repository.records[event.EventID+":region_created"]
+	openDue := repository.records[event.EventID+":open_due"]
+	if roomCreated.GroupID != "9001" || roomCreated.Scope != broadcastdomain.ScopeRoom || roomCreated.NextRetryAtMS != nowMS {
+		t.Fatalf("room created outbox mismatch: %+v", roomCreated)
+	}
+	if regionCreated.GroupID != "hy_lalu_bc_r_310" || regionCreated.Scope != broadcastdomain.ScopeRegion || regionCreated.NextRetryAtMS != nowMS {
+		t.Fatalf("region created outbox mismatch: %+v", regionCreated)
+	}
+	if openDue.GroupID != "9001" || openDue.Scope != broadcastdomain.ScopeRoom || openDue.NextRetryAtMS != openAtMS {
+		t.Fatalf("open due outbox mismatch: %+v", openDue)
+	}
+	regionData := redPacketPayloadData(t, regionCreated.PayloadJSON)
+	if regionData["event"] != "CREATED" || regionData["scope"] != "REGION" || regionData["packetMode"] != "DELAYED" {
+		t.Fatalf("region created payload mismatch: %+v", regionData)
+	}
+	openDueData := redPacketPayloadData(t, openDue.PayloadJSON)
+	if openDueData["event"] != "OPEN_DUE" || openDueData["scope"] != "ROOM" || openDueData["claimStartTime"] != float64(openAtMS) {
+		t.Fatalf("open due payload mismatch: %+v", openDueData)
+	}
+
+	result, err := service.ProcessPendingBroadcasts(appcode.WithContext(context.Background(), "lalu"), WorkerOptions{WorkerID: "worker-a", BatchSize: 10})
+	if err != nil {
+		t.Fatalf("ProcessPendingBroadcasts before open failed: %v", err)
+	}
+	if result.SuccessCount != 2 || len(publisher.messages) != 2 || containsCustomMessageEvent(publisher.messages, event.EventID+":open_due") {
+		t.Fatalf("before open only room/region CREATED should be sent: result=%+v messages=%+v", result, publisher.messages)
+	}
+
+	service.SetClock(func() time.Time { return time.UnixMilli(openAtMS) })
+	result, err = service.ProcessPendingBroadcasts(appcode.WithContext(context.Background(), "lalu"), WorkerOptions{WorkerID: "worker-a", BatchSize: 10})
+	if err != nil {
+		t.Fatalf("ProcessPendingBroadcasts after open failed: %v", err)
+	}
+	if result.SuccessCount != 1 || !containsCustomMessageEvent(publisher.messages, event.EventID+":open_due") {
+		t.Fatalf("after open should send OPEN_DUE: result=%+v messages=%+v", result, publisher.messages)
+	}
+}
+
 func TestRemoveRegionBroadcastMemberDeletesOnlyUserMembership(t *testing.T) {
 	publisher := &fakePublisher{}
 	service := New(Config{NodeID: "node-a"}, newFakeRepository(), publisher, nil)
@ -211,6 +287,15 @@ func redPacketPayloadData(t *testing.T, payloadJSON string) map[string]any {
 	return payload.Data
 }

+func containsCustomMessageEvent(messages []tencentim.CustomGroupMessage, eventID string) bool {
+	for _, message := range messages {
+		if message.EventID == eventID {
+			return true
+		}
+	}
+	return false
+}
+
 type fakeRepository struct {
 	records map[string]broadcastdomain.OutboxRecord
 }
@ -227,7 +312,7 @@ func (r *fakeRepository) SaveBroadcastOutbox(_ context.Context, record broadcast
 	return true, record, nil
 }

-func (r *fakeRepository) ClaimPendingBroadcastOutbox(_ context.Context, workerID string, _ int64, lockUntilMS int64, limit int, _ int) ([]broadcastdomain.OutboxRecord, error) {
+func (r *fakeRepository) ClaimPendingBroadcastOutbox(_ context.Context, workerID string, nowMS int64, lockUntilMS int64, limit int, _ int) ([]broadcastdomain.OutboxRecord, error) {
 	records := make([]broadcastdomain.OutboxRecord, 0, limit)
 	for eventID, record := range r.records {
 		if len(records) >= limit {
@ -236,6 +321,9 @@ func (r *fakeRepository) ClaimPendingBroadcastOutbox(_ context.Context, workerID
 		if record.Status != broadcastdomain.StatusPending && record.Status != broadcastdomain.StatusRetryable {
 			continue
 		}
+		if record.NextRetryAtMS > nowMS {
+			continue
+		}
 		record.Status = broadcastdomain.StatusDelivering
 		record.LockedBy = workerID
 		record.LockedUntilMS = lockUntilMS
--- a/services/gateway-service/internal/transport/http/response_test.go
+++ b/services/gateway-service/internal/transport/http/response_test.go
@ -4295,11 +4295,12 @@ func TestVoiceRoomActiveRedPacketRouteIsRegistered(t *testing.T) {
 	walletClient := &fakeWalletClient{listRedPacketsResp: &walletv1.ListRedPacketsResponse{
 		ServerTimeMs: 4000,
 		Packets: []*walletv1.RedPacket{
-			{PacketId: "packet-active", RoomId: "room-1", RegionId: 1001, PacketType: "normal", Status: "active"},
-			{PacketId: "packet-finished", RoomId: "room-1", RegionId: 1001, PacketType: "normal", Status: "finished"},
+			{PacketId: "packet-active", RoomId: "room-1", RegionId: 1001, SenderUserId: 10002, PacketType: "normal", Status: "active"},
+			{PacketId: "packet-finished", RoomId: "room-1", RegionId: 1001, SenderUserId: 10003, PacketType: "normal", Status: "finished"},
 		},
 	}}
-	handler := NewHandlerWithClients(&fakeRoomClient{}, nil, nil, &fakeUserProfileClient{regionID: 1001})
+	profileClient := &fakeUserProfileClient{regionID: 1001}
+	handler := NewHandlerWithClients(&fakeRoomClient{}, nil, nil, profileClient)
 	handler.SetWalletClient(walletClient)
 	router := handler.Routes(auth.NewVerifier("secret"))
 	request := httptest.NewRequest(http.MethodGet, "/api/v1/app/voice-room/red-packet/active?roomId=room-1", nil)
@ -4315,6 +4316,9 @@ func TestVoiceRoomActiveRedPacketRouteIsRegistered(t *testing.T) {
 	if walletClient.lastListRedPackets == nil || walletClient.lastListRedPackets.GetRoomId() != "room-1" || walletClient.lastListRedPackets.GetViewerUserId() != 42 {
 		t.Fatalf("active red-packet request mismatch: %+v", walletClient.lastListRedPackets)
 	}
+	if profileClient.lastBatch == nil || len(profileClient.lastBatch.GetUserIds()) != 1 || profileClient.lastBatch.GetUserIds()[0] != 10002 {
+		t.Fatalf("active red-packet should batch only visible sender profiles: %+v", profileClient.lastBatch)
+	}
 	var response httpkit.ResponseEnvelope
 	if err := json.NewDecoder(recorder.Body).Decode(&response); err != nil {
 		t.Fatalf("decode active red-packet response failed: %v", err)
@ -4325,7 +4329,7 @@ func TestVoiceRoomActiveRedPacketRouteIsRegistered(t *testing.T) {
 		t.Fatalf("active red-packet response mismatch: %+v", response)
 	}
 	first, ok := packets[0].(map[string]any)
-	if !ok || first["packetNo"] != "packet-active" || first["status"] != "ACTIVE" {
+	if !ok || first["packetNo"] != "packet-active" || first["status"] != "ACTIVE" || first["senderNickname"] != "user-10002" || first["senderAvatar"] != "https://cdn.example/avatar.png" {
 		t.Fatalf("active packet item mismatch: %+v", first)
 	}
 }
--- a/services/gateway-service/internal/transport/http/walletapi/red_packet_handler.go
+++ b/services/gateway-service/internal/transport/http/walletapi/red_packet_handler.go
@ -272,7 +272,7 @@ func (h *Handler) sendVoiceRoomRedPacket(writer http.ResponseWriter, request *ht
 }

 func (h *Handler) listActiveRedPackets(writer http.ResponseWriter, request *http.Request) {
-	if h.walletClient == nil {
+	if h.walletClient == nil || h.userProfileClient == nil {
 		httpkit.WriteError(writer, request, http.StatusBadGateway, httpkit.CodeUpstreamError, "upstream service error")
 		return
 	}
@ -295,12 +295,21 @@ func (h *Handler) listActiveRedPackets(writer http.ResponseWriter, request *http
 		writeRedPacketRPCError(writer, request, err)
 		return
 	}
-	packets := make([]map[string]any, 0, len(resp.GetPackets()))
+	activePackets := make([]*walletv1.RedPacket, 0, len(resp.GetPackets()))
 	for _, packet := range resp.GetPackets() {
 		if packet.GetStatus() == "active" || packet.GetStatus() == "waiting_open" {
-			packets = append(packets, voiceRoomPacketFromProto(packet))
+			activePackets = append(activePackets, packet)
 		}
 	}
+	profiles, err := h.redPacketSenderProfiles(request, activePackets)
+	if err != nil {
+		httpkit.WriteRPCError(writer, request, err)
+		return
+	}
+	packets := make([]map[string]any, 0, len(activePackets))
+	for _, packet := range activePackets {
+		packets = append(packets, voiceRoomPacketFromProtoWithSender(packet, profiles[packet.GetSenderUserId()]))
+	}
 	httpkit.WriteOK(writer, request, map[string]any{
 		"packets":    packets,
 		"serverTime": resp.GetServerTimeMs(),
@ -494,6 +503,34 @@ func (h *Handler) redPacketClaimProfiles(request *http.Request, claims []*wallet
 	return resp.GetUsers(), nil
 }

+func (h *Handler) redPacketSenderProfiles(request *http.Request, packets []*walletv1.RedPacket) (map[int64]*userv1.User, error) {
+	userIDs := make([]int64, 0, len(packets))
+	seen := make(map[int64]struct{}, len(packets))
+	for _, packet := range packets {
+		userID := packet.GetSenderUserId()
+		if userID <= 0 {
+			continue
+		}
+		if _, exists := seen[userID]; exists {
+			continue
+		}
+		seen[userID] = struct{}{}
+		userIDs = append(userIDs, userID)
+	}
+	if len(userIDs) == 0 {
+		return map[int64]*userv1.User{}, nil
+	}
+	// active 列表可能有多个红包，发送者展示资料必须批量拉取，避免房间侧边栏刷新造成 N+1 RPC。
+	resp, err := h.userProfileClient.BatchGetUsers(request.Context(), &userv1.BatchGetUsersRequest{
+		Meta:    httpkit.UserMeta(request, ""),
+		UserIds: userIDs,
+	})
+	if err != nil {
+		return nil, err
+	}
+	return resp.GetUsers(), nil
+}
+
 func (h *Handler) redPacketSenderProfile(request *http.Request, packet *walletv1.RedPacket, currentUser *userv1.User) (*userv1.User, error) {
 	senderUserID := packet.GetSenderUserId()
 	if senderUserID <= 0 {