From 6d35f421c0e304a3695610c027e7015f60835e08 Mon Sep 17 00:00:00 2001 From: zhx Date: Mon, 1 Jun 2026 19:42:53 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AD=97=E6=AE=B5=E8=A1=A5=E5=85=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/flutter对接/语音房红包Flutter对接.md | 5 +- .../service/broadcast/service_test.go | 90 ++++++++++++++++++- .../internal/transport/http/response_test.go | 12 ++- .../http/walletapi/red_packet_handler.go | 43 ++++++++- 4 files changed, 141 insertions(+), 9 deletions(-) diff --git a/docs/flutter对接/语音房红包Flutter对接.md b/docs/flutter对接/语音房红包Flutter对接.md index 8508376c..9e489280 100644 --- 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 最新查询结果为准。 diff --git a/services/activity-service/internal/service/broadcast/service_test.go b/services/activity-service/internal/service/broadcast/service_test.go index ed0c89b5..c6819010 100644 --- 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 diff --git a/services/gateway-service/internal/transport/http/response_test.go b/services/gateway-service/internal/transport/http/response_test.go index 8b4a7dd2..4e6ef505 100644 --- 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) } } diff --git a/services/gateway-service/internal/transport/http/walletapi/red_packet_handler.go b/services/gateway-service/internal/transport/http/walletapi/red_packet_handler.go index 8f38ab75..a0c8e625 100644 --- 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 {