hy/hyapp-server
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

广播增加用户信息

Browse Source
This commit is contained in:
zhx 2026-06-01 14:06:31 +08:00
parent 880f151feb
commit f76d2350b1
11 changed files with 516 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

87
services/activity-service/internal/app/app.go
View File

@ -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",

50
services/activity-service/internal/app/red_packet_event_test.go Normal file
View File

@ -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)
}
}

45
services/activity-service/internal/client/user_profile_source.go Normal file
View File

@ -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
}

38
services/activity-service/internal/domain/broadcast/broadcast.go
View File

@ -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.

62
services/activity-service/internal/service/broadcast/red_packet_wallet_outbox.go
View File

@ -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
}

35
services/activity-service/internal/service/broadcast/service.go
View File

@ -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) {

74
services/activity-service/internal/service/broadcast/service_test.go
View File

@ -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
}

57
services/gateway-service/internal/transport/http/response_test.go
View File

@ -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}},

58
services/gateway-service/internal/transport/http/walletapi/red_packet_handler.go
View File

@ -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(),

27
services/wallet-service/internal/storage/mysql/red_packet_repository.go
View File

@ -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,
}

36
services/wallet-service/internal/storage/mysql/red_packet_repository_test.go Normal file
View File

@ -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)
}
}