From f76d2350b1c546bd1a64f8affbd8557314c78be1 Mon Sep 17 00:00:00 2001 From: zhx Date: Mon, 1 Jun 2026 14:06:31 +0800 Subject: [PATCH] =?UTF-8?q?=E5=B9=BF=E6=92=AD=E5=A2=9E=E5=8A=A0=E7=94=A8?= =?UTF-8?q?=E6=88=B7=E4=BF=A1=E6=81=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- services/activity-service/internal/app/app.go | 87 +++++++++++++++---- .../internal/app/red_packet_event_test.go | 50 +++++++++++ .../internal/client/user_profile_source.go | 45 ++++++++++ .../internal/domain/broadcast/broadcast.go | 38 +++++--- .../broadcast/red_packet_wallet_outbox.go | 62 +++++++++++++ .../internal/service/broadcast/service.go | 35 ++++++-- .../service/broadcast/service_test.go | 74 ++++++++++++++++ .../internal/transport/http/response_test.go | 57 +++++++++++- .../http/walletapi/red_packet_handler.go | 58 ++++++++++++- .../storage/mysql/red_packet_repository.go | 27 +++--- .../mysql/red_packet_repository_test.go | 36 ++++++++ 11 files changed, 516 insertions(+), 53 deletions(-) create mode 100644 services/activity-service/internal/app/red_packet_event_test.go create mode 100644 services/activity-service/internal/client/user_profile_source.go create mode 100644 services/wallet-service/internal/storage/mysql/red_packet_repository_test.go diff --git a/services/activity-service/internal/app/app.go b/services/activity-service/internal/app/app.go index 894a1d7e..a75a8632 100644 --- a/services/activity-service/internal/app/app.go +++ b/services/activity-service/internal/app/app.go @@ -137,6 +137,7 @@ func New(cfg config.Config) (*App, error) { WorkerPollInterval: cfg.Broadcast.WorkerPollInterval, EnsureGroupsOnStartup: cfg.Broadcast.EnsureGroupsOnStartup, }, repository, broadcastPublisher, client.NewGRPCRegionSource(userConn)) + broadcastSvc.SetSenderProfileSource(client.NewGRPCUserProfileSource(userConn)) luckyGiftSvc := luckygiftservice.New(repository, luckygiftservice.WithWallet(walletv1.NewWalletServiceClient(walletConn)), luckygiftservice.WithRoomPublisher(tencentClient), @@ -444,25 +445,32 @@ func redPacketEventFromWalletMessage(body []byte) (broadcastdomain.RedPacketWall if message.EventType != "WalletRedPacketCreated" && message.EventType != "WalletRedPacketClaimed" && message.EventType != "WalletRedPacketRefunded" { return broadcastdomain.RedPacketWalletEvent{}, false, nil } - packetID, packetType, roomID, regionID, status, openAtMS, expiresAtMS := redPacketFieldsFromWalletPayload(message.PayloadJSON, message.EventType) - if packetID == "" { + fields := redPacketFieldsFromWalletPayload(message.PayloadJSON, message.EventType) + if fields.PacketID == "" { return broadcastdomain.RedPacketWalletEvent{}, false, nil } return broadcastdomain.RedPacketWalletEvent{ - AppCode: appcode.Normalize(message.AppCode), - EventID: message.EventID, - EventType: message.EventType, - TransactionID: message.TransactionID, - CommandID: message.CommandID, - PayloadJSON: message.PayloadJSON, - PacketID: packetID, - PacketType: packetType, - RoomID: roomID, - RegionID: regionID, - Status: status, - OpenAtMS: openAtMS, - ExpiresAtMS: expiresAtMS, - CreatedAtMS: message.OccurredAtMS, + AppCode: appcode.Normalize(message.AppCode), + EventID: message.EventID, + EventType: message.EventType, + TransactionID: message.TransactionID, + CommandID: message.CommandID, + PayloadJSON: message.PayloadJSON, + PacketID: fields.PacketID, + PacketType: fields.PacketType, + RoomID: fields.RoomID, + RegionID: fields.RegionID, + RegionCode: fields.RegionCode, + SenderUserID: fields.SenderUserID, + TotalAmount: fields.TotalAmount, + TotalCount: fields.TotalCount, + RemainingAmount: fields.RemainingAmount, + RemainingCount: fields.RemainingCount, + RefundedAmount: fields.RefundedAmount, + Status: fields.Status, + OpenAtMS: fields.OpenAtMS, + ExpiresAtMS: fields.ExpiresAtMS, + CreatedAtMS: message.OccurredAtMS, }, true, nil } @@ -487,10 +495,27 @@ func rechargeFactFromWalletPayload(payload string) (int64, string) { return sequence, rechargeType } -func redPacketFieldsFromWalletPayload(payload string, eventType string) (string, string, string, int64, string, int64, int64) { +type redPacketPayloadFields struct { + PacketID string + PacketType string + RoomID string + RegionID int64 + RegionCode string + SenderUserID int64 + TotalAmount int64 + TotalCount int32 + RemainingAmount int64 + RemainingCount int32 + RefundedAmount int64 + Status string + OpenAtMS int64 + ExpiresAtMS int64 +} + +func redPacketFieldsFromWalletPayload(payload string, eventType string) redPacketPayloadFields { var decoded map[string]any if err := json.Unmarshal([]byte(payload), &decoded); err != nil { - return "", "", "", 0, "", 0, 0 + return redPacketPayloadFields{} } packetID := stringFromDecoded(decoded, "packet_id") if packetID == "" { @@ -505,7 +530,27 @@ func redPacketFieldsFromWalletPayload(payload string, eventType string) (string, if eventType == "WalletRedPacketClaimed" || eventType == "WalletRedPacketRefunded" { status = stringFromDecoded(decoded, "packet_status") } - return packetID, packetType, roomID, regionID, status, openAtMS, expiresAtMS + totalCount := int32FromDecoded(decoded, "packet_count") + remainingCount := int32FromDecoded(decoded, "remaining_count") + if remainingCount == 0 && eventType == "WalletRedPacketCreated" { + remainingCount = totalCount + } + return redPacketPayloadFields{ + PacketID: packetID, + PacketType: packetType, + RoomID: roomID, + RegionID: regionID, + RegionCode: stringFromDecoded(decoded, "region_code"), + SenderUserID: int64FromDecoded(decoded, "sender_user_id"), + TotalAmount: int64FromDecoded(decoded, "total_amount"), + TotalCount: totalCount, + RemainingAmount: int64FromDecoded(decoded, "remaining_amount"), + RemainingCount: remainingCount, + RefundedAmount: int64FromDecoded(decoded, "refund_amount"), + Status: status, + OpenAtMS: openAtMS, + ExpiresAtMS: expiresAtMS, + } } func stringFromDecoded(decoded map[string]any, key string) string { @@ -527,6 +572,10 @@ func int64FromDecoded(decoded map[string]any, key string) int64 { } } +func int32FromDecoded(decoded map[string]any, key string) int32 { + return int32(int64FromDecoded(decoded, key)) +} + func luckyGiftWorkerOptions(nodeID string, cfg config.LuckyGiftWorkerConfig) luckygiftservice.WorkerOptions { return luckygiftservice.WorkerOptions{ WorkerID: nodeID + "-lucky-gift", diff --git a/services/activity-service/internal/app/red_packet_event_test.go b/services/activity-service/internal/app/red_packet_event_test.go new file mode 100644 index 00000000..ec49756d --- /dev/null +++ b/services/activity-service/internal/app/red_packet_event_test.go @@ -0,0 +1,50 @@ +package app + +import ( + "testing" + + "hyapp/pkg/walletmq" +) + +func TestRedPacketEventFromWalletMessageKeepsDisplayPayloadFacts(t *testing.T) { + body, err := walletmq.EncodeWalletOutboxMessage(walletmq.WalletOutboxMessage{ + AppCode: "lalu", + EventID: "wallet-red-packet-created-1", + EventType: "WalletRedPacketCreated", + TransactionID: "tx-red-packet-1", + CommandID: "cmd-red-packet-1", + UserID: 10001, + PayloadJSON: `{ + "packet_id":"VRRP2026052712340001", + "packet_type":"delayed", + "sender_user_id":10001, + "room_id":"9001", + "region_id":310, + "total_amount":15000, + "packet_count":5, + "remaining_amount":15000, + "remaining_count":5, + "status":"waiting_open", + "open_at_ms":1780000000000, + "expires_at_ms":1780086400000 + }`, + OccurredAtMS: 1_779_999_700_000, + }) + if err != nil { + t.Fatalf("encode wallet message failed: %v", err) + } + + event, ok, err := redPacketEventFromWalletMessage(body) + if err != nil || !ok { + t.Fatalf("decode red packet event failed: ok=%t err=%v", ok, err) + } + if event.PacketID != "VRRP2026052712340001" || event.PacketType != "delayed" || event.RoomID != "9001" || event.RegionID != 310 { + t.Fatalf("red packet identity fields mismatch: %+v", event) + } + if event.SenderUserID != 10001 || event.TotalAmount != 15000 || event.TotalCount != 5 || event.RemainingAmount != 15000 || event.RemainingCount != 5 { + t.Fatalf("red packet display facts mismatch: %+v", event) + } + if event.Status != "waiting_open" || event.OpenAtMS != 1_780_000_000_000 || event.ExpiresAtMS != 1_780_086_400_000 || event.CreatedAtMS != 1_779_999_700_000 { + t.Fatalf("red packet timing/status fields mismatch: %+v", event) + } +} diff --git a/services/activity-service/internal/client/user_profile_source.go b/services/activity-service/internal/client/user_profile_source.go new file mode 100644 index 00000000..2136b3e7 --- /dev/null +++ b/services/activity-service/internal/client/user_profile_source.go @@ -0,0 +1,45 @@ +package client + +import ( + "context" + + "google.golang.org/grpc" + userv1 "hyapp.local/api/proto/user/v1" + "hyapp/pkg/appcode" + broadcastservice "hyapp/services/activity-service/internal/service/broadcast" +) + +// GRPCUserProfileSource 从 user-service 读取红包播报所需的发送者展示资料。 +// activity-service 只消费当前快照组装 IM，不复制或拥有用户资料。 +type GRPCUserProfileSource struct { + client userv1.UserServiceClient +} + +func NewGRPCUserProfileSource(conn *grpc.ClientConn) *GRPCUserProfileSource { + return &GRPCUserProfileSource{client: userv1.NewUserServiceClient(conn)} +} + +func (s *GRPCUserProfileSource) GetSenderProfile(ctx context.Context, userID int64) (broadcastservice.SenderProfile, error) { + resp, err := s.client.GetUser(ctx, &userv1.GetUserRequest{ + Meta: &userv1.RequestMeta{ + RequestId: "red-packet-broadcast-profile", + Caller: "activity-service", + AppCode: appcode.FromContext(ctx), + }, + UserId: userID, + }) + if err != nil { + return broadcastservice.SenderProfile{}, err + } + user := resp.GetUser() + if user == nil { + return broadcastservice.SenderProfile{UserID: userID}, nil + } + return broadcastservice.SenderProfile{ + UserID: user.GetUserId(), + Account: user.GetDisplayUserId(), + Nickname: user.GetUsername(), + Avatar: user.GetAvatar(), + RegionCode: user.GetRegionCode(), + }, nil +} diff --git a/services/activity-service/internal/domain/broadcast/broadcast.go b/services/activity-service/internal/domain/broadcast/broadcast.go index 1117fe9a..79f98b81 100644 --- a/services/activity-service/internal/domain/broadcast/broadcast.go +++ b/services/activity-service/internal/domain/broadcast/broadcast.go @@ -78,20 +78,30 @@ type ConsumeRoomEventResult struct { // RedPacketWalletEvent 是 wallet-service 已提交的红包资金事实。 type RedPacketWalletEvent struct { - AppCode string - EventID string - EventType string - TransactionID string - CommandID string - PayloadJSON string - PacketID string - PacketType string - RoomID string - RegionID int64 - Status string - OpenAtMS int64 - ExpiresAtMS int64 - CreatedAtMS int64 + AppCode string + EventID string + EventType string + TransactionID string + CommandID string + PayloadJSON string + PacketID string + PacketType string + RoomID string + RegionID int64 + RegionCode string + SenderUserID int64 + SenderAccount string + SenderNickname string + SenderAvatar string + TotalAmount int64 + TotalCount int32 + RemainingAmount int64 + RemainingCount int32 + RefundedAmount int64 + Status string + OpenAtMS int64 + ExpiresAtMS int64 + CreatedAtMS int64 } // RedPacketCreatedEvent keeps older tests and callers compiling while wallet red-packet facts are generalized. diff --git a/services/activity-service/internal/service/broadcast/red_packet_wallet_outbox.go b/services/activity-service/internal/service/broadcast/red_packet_wallet_outbox.go index b962cc9b..a2065572 100644 --- a/services/activity-service/internal/service/broadcast/red_packet_wallet_outbox.go +++ b/services/activity-service/internal/service/broadcast/red_packet_wallet_outbox.go @@ -3,6 +3,7 @@ package broadcast import ( "context" "encoding/json" + "strconv" "strings" "hyapp/pkg/appcode" @@ -39,6 +40,11 @@ func (s *Service) consumeRedPacketCreated(ctx context.Context, event broadcastdo if event.RoomID == "" { return nil } + enriched, err := s.withRedPacketSenderProfile(ctx, event) + if err != nil { + return err + } + event = enriched if err := s.publishRedPacketRoomEvent(ctx, event, "CREATED", event.CreatedAtMS); err != nil { return err } @@ -65,6 +71,29 @@ func (s *Service) consumeRedPacketCreated(ctx context.Context, event broadcastdo return nil } +func (s *Service) withRedPacketSenderProfile(ctx context.Context, event broadcastdomain.RedPacketWalletEvent) (broadcastdomain.RedPacketWalletEvent, error) { + if event.SenderUserID <= 0 || s == nil || s.senderProfileSource == nil { + return event, nil + } + profile, err := s.senderProfileSource.GetSenderProfile(ctx, event.SenderUserID) + if err != nil { + return event, err + } + if event.SenderAccount == "" { + event.SenderAccount = profile.Account + } + if event.SenderNickname == "" { + event.SenderNickname = profile.Nickname + } + if event.SenderAvatar == "" { + event.SenderAvatar = profile.Avatar + } + if event.RegionCode == "" { + event.RegionCode = profile.RegionCode + } + return event, nil +} + func (s *Service) publishRedPacketRoomEvent(ctx context.Context, event broadcastdomain.RedPacketWalletEvent, eventName string, fallbackDeliverAtMS int64) error { if event.RoomID == "" { return nil @@ -94,10 +123,22 @@ func redPacketIMPayload(event broadcastdomain.RedPacketWalletEvent, eventName st "event": eventName, "scope": scope, "packetNo": event.PacketID, + "packetId": event.PacketID, "roomId": event.RoomID, "regionId": event.RegionID, + "regionCode": event.RegionCode, "packetMode": packetMode, + "senderUserId": redPacketUserIDString(event.SenderUserID), + "actualAccount": redPacketSenderAccount(event), + "userNickname": event.SenderNickname, + "userAvatar": event.SenderAvatar, + "totalAmount": event.TotalAmount, + "totalCount": event.TotalCount, + "remainAmount": event.RemainingAmount, + "remainCount": event.RemainingCount, "status": status, + "delaySeconds": redPacketDelaySeconds(event), + "refundedAmount": event.RefundedAmount, "claimStartTime": event.OpenAtMS, "expireTime": event.ExpiresAtMS, "serverTime": event.CreatedAtMS, @@ -109,3 +150,24 @@ func redPacketIMPayload(event broadcastdomain.RedPacketWalletEvent, eventName st } return string(encoded) } + +func redPacketUserIDString(userID int64) string { + if userID <= 0 { + return "" + } + return strconv.FormatInt(userID, 10) +} + +func redPacketSenderAccount(event broadcastdomain.RedPacketWalletEvent) string { + if strings.TrimSpace(event.SenderAccount) != "" { + return event.SenderAccount + } + return redPacketUserIDString(event.SenderUserID) +} + +func redPacketDelaySeconds(event broadcastdomain.RedPacketWalletEvent) int64 { + if !strings.EqualFold(event.PacketType, "delayed") || event.OpenAtMS <= event.CreatedAtMS { + return 0 + } + return (event.OpenAtMS - event.CreatedAtMS) / 1000 +} diff --git a/services/activity-service/internal/service/broadcast/service.go b/services/activity-service/internal/service/broadcast/service.go index e69f2bd6..103f71fa 100644 --- a/services/activity-service/internal/service/broadcast/service.go +++ b/services/activity-service/internal/service/broadcast/service.go @@ -46,6 +46,22 @@ type RegionSource interface { ListActiveRegionIDs(ctx context.Context) ([]int64, error) } +// SenderProfileSource 只读取红包 IM 展示所需的公开资料快照。 +// 资金和红包状态仍来自 wallet outbox，用户资料 owner 仍是 user-service。 +type SenderProfileSource interface { + GetSenderProfile(ctx context.Context, userID int64) (SenderProfile, error) +} + +// SenderProfile 是红包 IM payload 的发送者展示字段。 +// activity-service 不持久化该资料，只在消费 wallet 事实时组装一次播报消息。 +type SenderProfile struct { + UserID int64 + Account string + Nickname string + Avatar string + RegionCode string +} + // Config 保存播报策略和 worker 参数。 // SuperGiftMinValue 是产品策略阈值；Worker* 是 outbox 投递吞吐和故障恢复参数。 type Config struct { @@ -71,11 +87,12 @@ type PublishInput struct { // Service 拥有播报群建群补偿、播报入队和 outbox 投递决策。 // 它不直接向客户端返回 UI 文案，只生产带 action 的结构化 JSON，让客户端自行决定展示形态。 type Service struct { - cfg Config - repository Repository - publisher Publisher - regionSource RegionSource - now func() time.Time + cfg Config + repository Repository + publisher Publisher + regionSource RegionSource + senderProfileSource SenderProfileSource + now func() time.Time } // New 创建播报服务。 @@ -92,6 +109,14 @@ func (s *Service) SetClock(now func() time.Time) { } } +// SetSenderProfileSource 注入 user-service 资料读取边界。 +// 测试或离线环境可以不注入，此时红包 IM payload 仍会带 user_id/account fallback。 +func (s *Service) SetSenderProfileSource(source SenderProfileSource) { + if s != nil { + s.senderProfileSource = source + } +} + // EnsureBroadcastGroups 幂等创建全局播报群和所有 active 区域播报群。 // 这个动作放在启动/定时 reconciler，不放在 /im/usersig 路径，避免用户登录被外部腾讯 REST 拖慢。 func (s *Service) EnsureBroadcastGroups(ctx context.Context) (int, error) { diff --git a/services/activity-service/internal/service/broadcast/service_test.go b/services/activity-service/internal/service/broadcast/service_test.go index ce9a8c15..ed0c89b5 100644 --- a/services/activity-service/internal/service/broadcast/service_test.go +++ b/services/activity-service/internal/service/broadcast/service_test.go @@ -105,6 +105,55 @@ func TestProcessPendingBroadcastsPublishesCustomMessage(t *testing.T) { } } +func TestConsumeRedPacketCreatedBuildsFullIMPayload(t *testing.T) { + repository := newFakeRepository() + service := New(Config{NodeID: "node-a"}, repository, nil, nil) + 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: 1_780_000_000_000, + ExpiresAtMS: 1_780_086_400_000, + CreatedAtMS: 1_779_999_700_000, + } + + if err := service.ConsumeRedPacketWalletEvent(appcode.WithContext(context.Background(), "lalu"), event); err != nil { + t.Fatalf("ConsumeRedPacketWalletEvent failed: %v", err) + } + + record := repository.records[event.EventID] + data := redPacketPayloadData(t, record.PayloadJSON) + if data["packetNo"] != event.PacketID || data["packetId"] != event.PacketID || data["scope"] != "ROOM" { + t.Fatalf("packet identity mismatch: %+v", data) + } + if data["senderUserId"] != "10001" || data["actualAccount"] != "10001" || data["userNickname"] != "Tom" || data["userAvatar"] != "https://example.com/avatar.png" { + t.Fatalf("sender display fields mismatch: %+v", data) + } + if data["totalAmount"] != float64(15000) || data["totalCount"] != float64(5) || data["remainAmount"] != float64(15000) || data["remainCount"] != float64(5) { + t.Fatalf("red packet amount/count fields mismatch: %+v", data) + } + if data["packetMode"] != "DELAYED" || data["status"] != "WAITING_OPEN" || data["delaySeconds"] != float64(300) || data["regionCode"] != "AR" { + t.Fatalf("red packet display state mismatch: %+v", data) + } +} + func TestRemoveRegionBroadcastMemberDeletesOnlyUserMembership(t *testing.T) { publisher := &fakePublisher{} service := New(Config{NodeID: "node-a"}, newFakeRepository(), publisher, nil) @@ -146,6 +195,22 @@ func mustGiftEnvelope(t *testing.T, gift *roomeventsv1.RoomGiftSent) *roomevents } } +func redPacketPayloadData(t *testing.T, payloadJSON string) map[string]any { + t.Helper() + + var payload struct { + Type string `json:"type"` + Data map[string]any `json:"data"` + } + if err := json.Unmarshal([]byte(payloadJSON), &payload); err != nil { + t.Fatalf("payload is not json: %v", err) + } + if payload.Type != "ROOM_RED_PACKET" || payload.Data == nil { + t.Fatalf("unexpected red packet payload: %+v", payload) + } + return payload.Data +} + type fakeRepository struct { records map[string]broadcastdomain.OutboxRecord } @@ -225,3 +290,12 @@ func (p *fakePublisher) DeleteGroupMember(_ context.Context, groupID string, use p.deletedUserID = userID return nil } + +type fakeSenderProfileSource struct { + profile SenderProfile + err error +} + +func (s fakeSenderProfileSource) GetSenderProfile(context.Context, int64) (SenderProfile, error) { + return s.profile, s.err +} diff --git a/services/gateway-service/internal/transport/http/response_test.go b/services/gateway-service/internal/transport/http/response_test.go index 5a543f29..2e6ba792 100644 --- a/services/gateway-service/internal/transport/http/response_test.go +++ b/services/gateway-service/internal/transport/http/response_test.go @@ -457,6 +457,8 @@ type fakeWalletClient struct { lastGrantResource *walletv1.GrantResourceRequest grantResourceResp *walletv1.ResourceGrantResponse grantResourceErr error + lastGetRedPacket *walletv1.GetRedPacketRequest + getRedPacketResp *walletv1.GetRedPacketResponse } type fakeMessageInboxClient struct { @@ -1386,7 +1388,14 @@ func (f *fakeWalletClient) ListRedPackets(context.Context, *walletv1.ListRedPack return &walletv1.ListRedPacketsResponse{}, nil } -func (f *fakeWalletClient) GetRedPacket(context.Context, *walletv1.GetRedPacketRequest) (*walletv1.GetRedPacketResponse, error) { +func (f *fakeWalletClient) GetRedPacket(_ context.Context, req *walletv1.GetRedPacketRequest) (*walletv1.GetRedPacketResponse, error) { + f.lastGetRedPacket = req + if f.err != nil { + return nil, f.err + } + if f.getRedPacketResp != nil { + return f.getRedPacketResp, nil + } return &walletv1.GetRedPacketResponse{}, nil } @@ -4021,6 +4030,52 @@ func TestBatchUserProfilesReturnsOnlySafeDisplayFields(t *testing.T) { } } +func TestVoiceRoomRedPacketClaimRecordsIncludesUserProfiles(t *testing.T) { + walletClient := &fakeWalletClient{getRedPacketResp: &walletv1.GetRedPacketResponse{ + ServerTimeMs: 3000, + Packet: &walletv1.RedPacket{ + PacketId: "packet-1", + Claims: []*walletv1.RedPacketClaim{ + {ClaimId: "claim-1", PacketId: "packet-1", UserId: 10002, Amount: 300, Status: "claimed", CreatedAtMs: 1000}, + {ClaimId: "claim-2", PacketId: "packet-1", UserId: 10003, Amount: 200, Status: "claimed", CreatedAtMs: 2000}, + }, + }, + }} + profileClient := &fakeUserProfileClient{} + 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/claim-records?packetNo=packet-1&page=1&pageSize=1", nil) + request.Header.Set("Authorization", "Bearer "+signGatewayToken(t, "secret", 42)) + request.Header.Set("X-Request-ID", "req-red-packet-claims") + recorder := httptest.NewRecorder() + + router.ServeHTTP(recorder, request) + + if recorder.Code != http.StatusOK { + t.Fatalf("status mismatch: got %d body=%s", recorder.Code, recorder.Body.String()) + } + if walletClient.lastGetRedPacket == nil || !walletClient.lastGetRedPacket.GetIncludeClaims() || walletClient.lastGetRedPacket.GetPacketId() != "packet-1" { + t.Fatalf("red packet detail request mismatch: %+v", walletClient.lastGetRedPacket) + } + if profileClient.lastBatch == nil || len(profileClient.lastBatch.GetUserIds()) != 1 || profileClient.lastBatch.GetUserIds()[0] != 10002 { + t.Fatalf("claim records must batch only visible claim users: %+v", profileClient.lastBatch) + } + var response httpkit.ResponseEnvelope + if err := json.NewDecoder(recorder.Body).Decode(&response); err != nil { + t.Fatalf("decode claim records failed: %v", err) + } + data, ok := response.Data.(map[string]any) + items, itemsOK := data["items"].([]any) + if response.Code != httpkit.CodeOK || !ok || !itemsOK || len(items) != 1 { + t.Fatalf("claim records response mismatch: %+v", response) + } + first, ok := items[0].(map[string]any) + if !ok || first["account"] != "110002" || first["nickname"] != "user-10002" || first["avatar"] != "https://cdn.example/avatar.png" { + t.Fatalf("claim record should include display profile fields: %+v", first) + } +} + func TestGetMyBalancesUsesAuthenticatedUserIDAndDefaultsCoin(t *testing.T) { walletClient := &fakeWalletClient{resp: &walletv1.GetBalancesResponse{ Balances: []*walletv1.AssetBalance{{AssetType: "COIN", AvailableAmount: 120, FrozenAmount: 3, Version: 2}}, 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 174207e5..1a5aed08 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 @@ -358,7 +358,7 @@ func (h *Handler) claimVoiceRoomRedPacket(writer http.ResponseWriter, request *h return } httpkit.WriteOK(writer, request, map[string]any{ - "claim": voiceRoomClaimFromProto(resp.GetClaim()), + "claim": voiceRoomClaimFromProtoWithUser(resp.GetClaim(), user), "packet": voiceRoomPacketFromProto(resp.GetPacket()), "balanceAfter": resp.GetBalanceAfter(), "serverTime": resp.GetServerTimeMs(), @@ -395,6 +395,10 @@ func (h *Handler) getVoiceRoomRedPacket(writer http.ResponseWriter, request *htt } func (h *Handler) listRedPacketClaims(writer http.ResponseWriter, request *http.Request) { + if h.walletClient == nil || h.userProfileClient == nil { + httpkit.WriteError(writer, request, http.StatusBadGateway, httpkit.CodeUpstreamError, "upstream service error") + return + } packetNo := strings.TrimSpace(request.URL.Query().Get("packetNo")) if packetNo == "" { packetNo = strings.TrimSpace(request.URL.Query().Get("packet_id")) @@ -435,9 +439,14 @@ func (h *Handler) listRedPacketClaims(writer http.ResponseWriter, request *http. if end > len(all) { end = len(all) } + profiles, err := h.redPacketClaimProfiles(request, all[start:end]) + if err != nil { + httpkit.WriteRPCError(writer, request, err) + return + } items := make([]map[string]any, 0, end-start) for _, claim := range all[start:end] { - items = append(items, voiceRoomClaimFromProto(claim)) + items = append(items, voiceRoomClaimFromProtoWithUser(claim, profiles[claim.GetUserId()])) } httpkit.WriteOK(writer, request, map[string]any{ "items": items, @@ -448,6 +457,33 @@ func (h *Handler) listRedPacketClaims(writer http.ResponseWriter, request *http. }) } +func (h *Handler) redPacketClaimProfiles(request *http.Request, claims []*walletv1.RedPacketClaim) (map[int64]*userv1.User, error) { + userIDs := make([]int64, 0, len(claims)) + seen := make(map[int64]struct{}, len(claims)) + for _, claim := range claims { + userID := claim.GetUserId() + 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 + } + 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) listSentRedPackets(writer http.ResponseWriter, request *http.Request) { if h.walletClient == nil { httpkit.WriteError(writer, request, http.StatusBadGateway, httpkit.CodeUpstreamError, "upstream service error") @@ -903,13 +939,31 @@ func voiceRoomPacketFromProto(packet *walletv1.RedPacket) map[string]any { } func voiceRoomClaimFromProto(claim *walletv1.RedPacketClaim) map[string]any { + return voiceRoomClaimFromProtoWithUser(claim, nil) +} + +func voiceRoomClaimFromProtoWithUser(claim *walletv1.RedPacketClaim, user *userv1.User) map[string]any { if claim == nil { return map[string]any{} } + account := "" + nickname := "" + avatar := "" + if user != nil { + account = user.GetDisplayUserId() + nickname = user.GetUsername() + avatar = user.GetAvatar() + } + if account == "" && claim.GetUserId() > 0 { + account = strconv.FormatInt(claim.GetUserId(), 10) + } return map[string]any{ "claimId": claim.GetClaimId(), "packetNo": claim.GetPacketId(), "userId": claim.GetUserId(), + "account": account, + "nickname": nickname, + "avatar": avatar, "amount": claim.GetAmount(), "status": claim.GetStatus(), "createdAt": claim.GetCreatedAtMs(), diff --git a/services/wallet-service/internal/storage/mysql/red_packet_repository.go b/services/wallet-service/internal/storage/mysql/red_packet_repository.go index 8cd0d2f2..1ef2eec7 100644 --- a/services/wallet-service/internal/storage/mysql/red_packet_repository.go +++ b/services/wallet-service/internal/storage/mysql/red_packet_repository.go @@ -1231,18 +1231,21 @@ func redPacketCreatedEvent(transactionID string, commandID string, metadata redP AvailableDelta: -metadata.TotalAmount, FrozenDelta: 0, Payload: map[string]any{ - "transaction_id": transactionID, - "command_id": commandID, - "packet_id": metadata.PacketID, - "packet_type": metadata.PacketType, - "sender_user_id": metadata.SenderUserID, - "room_id": metadata.RoomID, - "region_id": metadata.RegionID, - "total_amount": metadata.TotalAmount, - "packet_count": metadata.PacketCount, - "open_at_ms": metadata.OpenAtMS, - "expires_at_ms": metadata.ExpiresAtMS, - "created_at_ms": metadata.CreatedAtMS, + "transaction_id": transactionID, + "command_id": commandID, + "packet_id": metadata.PacketID, + "packet_type": metadata.PacketType, + "sender_user_id": metadata.SenderUserID, + "room_id": metadata.RoomID, + "region_id": metadata.RegionID, + "total_amount": metadata.TotalAmount, + "packet_count": metadata.PacketCount, + "remaining_amount": metadata.RemainingAmount, + "remaining_count": metadata.RemainingCount, + "status": metadata.Status, + "open_at_ms": metadata.OpenAtMS, + "expires_at_ms": metadata.ExpiresAtMS, + "created_at_ms": metadata.CreatedAtMS, }, CreatedAtMS: nowMS, } diff --git a/services/wallet-service/internal/storage/mysql/red_packet_repository_test.go b/services/wallet-service/internal/storage/mysql/red_packet_repository_test.go new file mode 100644 index 00000000..a38c556e --- /dev/null +++ b/services/wallet-service/internal/storage/mysql/red_packet_repository_test.go @@ -0,0 +1,36 @@ +package mysql + +import ( + "testing" + + "hyapp/services/wallet-service/internal/domain/ledger" +) + +func TestRedPacketCreatedEventCarriesDisplayFacts(t *testing.T) { + event := redPacketCreatedEvent("tx-red-packet-1", "cmd-red-packet-1", redPacketMetadata{ + PacketID: "VRRP2026052712340001", + PacketType: ledger.RedPacketTypeDelayed, + SenderUserID: 10001, + RoomID: "9001", + RegionID: 310, + TotalAmount: 15000, + PacketCount: 5, + RemainingAmount: 15000, + RemainingCount: 5, + Status: ledger.RedPacketStatusWaitingOpen, + OpenAtMS: 1_780_000_000_000, + ExpiresAtMS: 1_780_086_400_000, + CreatedAtMS: 1_779_999_700_000, + }, 1_779_999_700_000) + + payload, ok := event.Payload.(map[string]any) + if !ok { + t.Fatalf("red packet created payload type mismatch: %T", event.Payload) + } + if payload["total_amount"] != int64(15000) || payload["packet_count"] != int32(5) || payload["remaining_count"] != int32(5) { + t.Fatalf("red packet created payload must carry amount/count facts: %+v", payload) + } + if payload["status"] != ledger.RedPacketStatusWaitingOpen || payload["sender_user_id"] != int64(10001) { + t.Fatalf("red packet created payload must carry status and sender facts: %+v", payload) + } +}